Python Revisited 2024

Introduction

A quick revisit to python to maybe improve/refresh my own knowledge.

Project Creation (Poetry)

Guess most folks know about this and knowing IT, some people will hate it, and some will love it. Coming fresh out of NodeJS and React/Nextjs, this seemed a good idea to me. I am starting to feel with AI, that technologies are going to be changing quickly and being able to get going quickly is useful. This looked easy but when I tried it, it was harder. Here goes

Make a Project

poetry new my-project

Modify the pyproject.toml

The main thing is to change the packages line. I had a bit of trouble getting pylance to find my packages because, maybe, I did not poetry install after this. See below.

[tool.poetry]
name = "my-project"
version = "0.1.0"
description = ""
authors = ["Bill Wiseman <bw@bibble.co.nz>"]
readme = "README.md"

packages = [{ include = "app", from = "src" }]

[tool.poetry.dependencies]
python = "^3.12"

[tool.poetry.group.dev.dependencies]

[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"

Flask

Create the Flask App

Now lets make a Flask app. We will start by create files in src/app

cd my-project
mkdir -p src/app
touch src/app/__init__.py
touch src/app/__main__.py

Next, Install Flask add run install poetry (npm i for poetry)

poetry add Flask
poetry install

Now we can add the content to src/app/__main__.py

from flask import Flask

def main() -> Flask:
    app = Flask(__name__)

    return app

def run():
    app = main()
    app.run(debug=True, host="localhost", port=5000)

if __name__ == "__main__":
    run()

You should now be able to run the application with

poetry run python3 -m  app

When I ran the code it picked up a previous project and I had to remove the cache with

rm -rf ~/.cache/pypoetry/virtualenvs/

Flask Framework

Clearly there is a reason for Flask. Hopefully this will demonstrate how it helps. We now modify the main() to be a bit more fleshed out. Each part I will try to explain. It is not there way but at least it shows an approach to use. For me, good approaches, or any approach, means that you can extend and use other people to help

New Main

def main() -> Flask:
    # Load the app configuration
    config = load_app_config()
    
    # Configure logging
    configure_logging(config.app_config)

    # Create a session maker for SQLAlchemy
    session_maker = create_session_maker(config.db_config.full_url)

    # Create a Flask app instance
    app = Flask(__name__)

    # Set up middleware 
    middlewares.register(app, session_maker)

    # Register the blueprints
    routes.register(app)
    errors.register(app)

    return app

def run():
    app = main()
    app.run(debug=True, host="localhost", port=5000)

if __name__ == "__main__":
    run()

Load Config

So clearly this is where you load the config. For me I chose to use environment variables but there is an example of config files in [here]

import os

from app.confguration.app_config.models import AppConfig, Config, DatabaseConfig

def load_app_config() -> Config:
    application_config = AppConfig(
        debug=True
    )   

    database_config = DatabaseConfig(
        account=os.getenv("SNOWFLAKE_ACCOUNT"),
        user_name=os.getenv("SNOWFLAKE_USERNAME"),
        warehouse=os.getenv("SNOWFLAKE_WAREHOUSE"),
        role=os.getenv("SNOWFLAKE_ROLE"),
        database=os.getenv("SNOWFLAKE_DATABASE"),
        schema=os.getenv("SNOWFLAKE_SCHEMA"),
        password=os.getenv("SNOWFLAKE_PASSWORD")
    )

    return Config(application_config, database_config)

Configure Logging

Again this can be done in may ways. Here is one approach

import logging

from flask import Config

from app.confguration.logging_config.ConsoleFormatter import ConsoleFormatter

DEFAULT_LOGGING_LEVEL: int = logging.INFO

def configure_logging(config: Config) -> None:
    logging_level: int = logging.DEBUG if config.debug else DEFAULT_LOGGING_LEVEL

    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging_level)
    console_handler.setFormatter(ConsoleFormatter())

    logging.basicConfig(handlers=[console_handler], level=logging_level)

Session Maker (For SQLAlchemy)

This is presumably the way to share configuration across the application for SQLAlchemy

from sqlalchemy import create_engine
from sqlalchemy.orm.session import sessionmaker

def create_session_maker(database_url: str) -> sessionmaker:
    engine = create_engine(
        database_url,
        echo=True,
        pool_size=15,
        max_overflow=15,
        connect_args={
            "connect_timeout": 5,
        },
    )
    return sessionmaker(engine, autoflush=False, expire_on_commit=False)

Middleware

This is the import bit for me. Hopefully I can demonstrate, not only passing of the session, but authentication. Here is the session first.

Register

Here is where we set up the registering of the middleware

from flask import Flask
from sqlalchemy.orm.session import sessionmaker

from .database_middleware import DatabaseMiddleware


def register(app: Flask, session_maker: sessionmaker) -> None:
    DatabaseMiddleware(session_maker).register(app)

Session (SQLAlchemy)

Here is the implementation of the session for SQLAlchemy. So it calls the open and close functions based on the app.before_request and app.teardown_appcontext. I guess that there will be some routes which do not require the database which will have this overhead but otherwise nice and easy to understand.

from flask import Flask, g
from sqlalchemy.orm.session import sessionmaker

class DatabaseMiddleware:
    def __init__(self, session_maker: sessionmaker):
        self.session_maker = session_maker

    def open(self):
        session = self.session_maker()
        g.session = session

    # noinspection PyMethodMayBeStatic
    def close(self, *_args, **_kwargs):
        g.session.close()

    def register(self, app: Flask):
        app.before_request(self.open)
        app.teardown_appcontext(self.close)

Blueprints

Errors

This seems to me way or organize you code more than anything else. For me I separated out the errors handling from the rest. In the __init__.py you define the register function

from flask import Flask
from .errors import error_blueprint

def register(app: Flask) -> None:
    app.register_blueprint(error_blueprint)

Then define the blueprint.

from flask import Blueprint, render_template


error_blueprint = Blueprint('errors', __name__, template_folder='templates')

@error_blueprint.app_errorhandler(404)
def handle_404(err):
    return render_template('404.html'), 404

@error_blueprint.app_errorhandler(500)
def handle_500(err):
    return render_template('500.html'), 500

The example shows how to put templates in the same structure

Header Files In Python

I have found it a real struggle to keep this in my mind, maybe because of dyslexia or just it is odd. So I write it here to help. Basically the __init__.py is the .def, .h or even the d.ts. If you see

TypeError: 'module' object is not callable

This probably means you have not export the function or type but have used it in the code. There seems to be a naming convention where they do not use the function name as the name of the file. For example flask-template uses session.py for the function create_session_maker. There is only one function in it so I don't understand this approach.

Removing Pesky pycache

To remove these just a this to your .vscode/settings.json

{
    "files.exclude": {
        "**/*.pyc": {"when": "$(basename).py"},
        "**/__pycache__": true
    },
}

Database

Native Approach

Obviously you need data, we used, for better or worse, snowflake. Originally I downloaded the snowflake-connector-python followed the documentation and it seemed to work.

 class RepositorySnowflake():
     def __init__(self, config):
         self.config = config
         self.connection = snowflake.connector.connect(
             user=config["SNOWFLAKE_USERNAME"],
             password=config["SNOWFLAKE_PASSWORD"],
             account=config["SNOWFLAKE_ACCOUNT"],
             warehouse=config["SNOWFLAKE_WAREHOUSE"],
             database=config["SNOWFLAKE_DATABASE"],
             schema=config["SNOWFLAKE_SCHEMA"]
         )
 
     def query(self, sql):
         cursor = self.connection.cursor()
         cursor.execute(sql)
         return cursor.fetchall()
     
     def close(self):
         self.connection.close()

This all worked and was pleased with the time it took to get going. Switching languages sometimes can take time for the little grey cells to click in.

SQLAlchemy Approach

This does seem to be the approach to take nowadays

SQLAlchemylchemy Error

I did have some problems getting sqlalchemy to work. It was a simple error but worth noting here to make sure I do not forget next time arround

Not an executable object: 'select current_version()'

This is caused by not surrounding the query by text()

    # Create Engine
    engine = create_engine(config.db_config.full_url)  

    try:
        connection = engine.connect()
        # error results = connection.execute("select current_version()").fetchone()
        results = connection.execute(text("select current_version()")).fetchone()
        print(results)
    except exc.SQLAlchemyError as e:
        print(e)
    finally:
        connection.close()
        engine.dispose()

SQLAlchemy Model

The model is an class which is used to map the results from a SQL statement to an instance of a class.

class CarModel(BaseModel):
    __tablename__ = "car"

    id: Mapped[int] = mapped_column(primary_key=True)
    rego: Mapped[str | None]
    color: Mapped[str | None]

Some of the examples you where it looks like this. This is the old approach for SQLAlchemy

class CarModel(BaseModel):
    __tablename__ = "car"

    id = Column(Integer,primary_key=True)
    rego = Column(String)
    color = Column(String)

pydantic Schema (Zod for Python)

The Schema is like a Zod schema and is used to validate the data in the model. Note the BaseModel is from pydantic in this case not sqlalchemy. Below I have used the @field_serializer to override the value of rego. If it is null, it is converted to empty string. I am sure there is a better way to do this but this worked at the time.

from pydantic import BaseModel, ConfigDict

class CarSchema(BaseModel):

    id: int
    rego: str | None
    color: str | None

    @field_serializer('rego')
    def none_to_empty(v: str) -> Optional[str]:
        if v == None:
            return ''
        return v

class CarSchema(CarSchemaCreate):

    id: int

    model_config = ConfigDict(from_attributes=True)

SQLAlchemy execute the SQL

Not a big fan of ORMS in general and it is because I have mostly worked without them. My hope is that they do hide the technology for those who do use them, but for me I have spent so much time figuring out how to work around them when they don't work. Anyway here is how to execute you statement once you have a

• session
• model
• schema

Using the orm, you can Order By, add Where Clause etc easily.

class CarController(Controller[CarModel]):

    def list_cars(self, id=None) -> list[CarSchema]:
        stmt = select(CarModel)

        # if id is not None, filter by id
        if id:
            stmt = stmt.where(CarModel.id == id)

        result = self.session.scalars(stmt.order_by(CarModel.id)).fetchall()
        return TypeAdapter(list[CarSchema]).validate_python(result)

SQLAlchemy execute the RAW SQL (with SQLAlchemy Model)

One of the things I wanted to do was to map my own SQL Statement which had complicated statements, e.g. partitions, to a SQLAlchemy model so it could be processed like any other SQLAlchemy model once the statement had been executed.

To do this I could just use a model as before. I did have to put a table name in __tablename__, but it does not have to exist. I think it just needs to be unique (maybe registered somewhere).
From there the model could be the same as before (with a different name)

class PseudoCarModel(BaseModel):
    __tablename__ = "pseudo_car"

    id: Mapped[int] = mapped_column(primary_key=True)
    rego: Mapped[str | None]
    color: Mapped[str | None]

For the execution of the statement we need to just change it slightly

        # Using generated statement
        # result = self.session.scalars(stmt.order_by(CarModel.vehicle_id)).fetchall()
        result = self.session.execute(select(PseudoCarModel).from_statement(text('SELECT blah from blah'))).scalars().all()
        return TypeAdapter(list[CarSchema]).validate_python(result)

SQLAlchemy Snowflake

This is a snowflake specific thing for how a company I was at used Snowflake. Instead of using Schemas a templates, they used them as if they were databases. So here is how you can workaround this issue

class VehicleDriver(BaseModel):
    __tablename__ = "MY_TABLE"
    __table_args__ = {
      'schema' : 'MYDB.MY_SCHEMA'
    }