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
Make a Fask App
Now lets make a Flask app. We will start by putting the code in src/app
cd my-project
mkdir -p src/app
touch src/app/__init__.py
touch src/app/__main__.py
Now we can make the 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()
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"
Install Flask
So we install Flask add run install poetry
poetry add Flask
poetry install
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/
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
sqlalchemy 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 snowflake
To set the schema for snowflake you can do the following.
class VehicleDriver(BaseModel):
__tablename__ = "MY_TABLE"
__table_args__ = {
'schema' : 'MYDB.MY_SCHEMA'
}
Model
The model is mapped from the SQL statement. Whilst you put the __tablename__, it can refer to a table that does not exist when you use your own SQL statement
class CarModel(BaseModel):
__tablename__ = "car"
id: Mapped[int] = mapped_column(primary_key=True)
rego: Mapped[str | None]
color: Mapped[str | None]
Schema
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
from pydantic import BaseModel
class CarSchema(BaseModel):
id: int
rego: str | None
color: str | None
You can use your own SQL in SQLAlchemy like this
result = self.session.execute(select(CarModel).from_statement(text('SELECT blah from blah'))).scalars().all()
return TypeAdapter(list[CarSchema]).validate_python(result)
pydantic (Zod for python)
This has two uses. It validates you schema against your model and you can customize the serialization. So given a sqlalchemy Model
from sqlalchemy.orm import Mapped, mapped_column
from .base import CreatedUpdatedAtMixin
class User(CreatedUpdatedAtMixin):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
username: Mapped[str | None] = mapped_column(unique=True)
first_name: Mapped[str]
last_name: Mapped[str]
This can be validated against a pydantic schema
from pydantic import BaseModel, ConfigDict
class UserCreate(BaseModel):
username: str | None
first_name: str
last_name: str
class User(UserCreate):
id: int
model_config = ConfigDict(from_attributes=True)
class UserUpdate(UserCreate):
pass
And you can change the serialization to change the name of a field or change the value.
@field_serializer('first_name')
def none_to_empty(v: str) -> Optional[str]:
if v == None:
return ''
return v