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
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()
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)
</syntaxhighlist>
==Session Maker==
==Middleware==
==Blueprints==
=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
<syntaxhighlight lang="py">
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'
}