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

New Main

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

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

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

    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()

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'
    }