Set up flask with SQLAlchemy

Integrates SQLAlchemy with Flask

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
 
app: Flask = Flask(__name__)
db: SQLAlchemy = SQLAlchemy()
# The import of MySQLAlchemyConfiguration must be after the initialization of db (SQLAlchemy instance) 
# because MySQLAlchemyConfiguration refers to db
import MySQLAlchemyConfiguration
# The init() function sets up our tables/models/repository
MySQLAlchemyConfiguration.init()

MySQLAlchemyConfiguration file:

from PersonRepository import PersonRepository
from flask_example import db, app
 
person_repository: PersonRepository
 
 
def init():
    global person_repository
    # configure the SQLite database, relative to the app instance folder
    app.config["SQLALCHEMY_DATABASE_URI"] = 'postgresql+psycopg2://postgres:secret@localhost/flask_db'
    # initialize the app with the extension
    db.init_app(app)
    with app.app_context():
        db.create_all()
    person_repository = PersonRepository()

Models and Tables

Generalities

Subclass db.Model class to define models, or the db.Table class to create tables.
Both handle Flask-SQLAlchemy’s bind keys to associate with a specific engine.

Generate tables from model

Use create_all() to create the models and tables after defining them:

with app.app_context():
    db.create_all()

Beware:
create_all() doesn’t update existing tables.
To update them you have to do that manually or with a script outside SQLAlchemy.

ORM model static definitions

Specify relationship of models

The relationship.backref keyword is deprecated, relationship.back_populates with explicit relationship() constructs should be preferred.
Using individual relationship() constructs provides advantages including that both ORM mapped classes will include their attributes up front as the class is constructed, rather than as a deferred step, and configuration is more straightforward as all arguments are explicit.

Static configuration: loading Relationships with the attribute lazy

relationship() as an attribute lazy that defines how the relationship will be loaded.
Default value is select.

Here are the possible values:

select: loaded lazily when the property is first accessed, using a separate SELECT statement, or identity map fetch for simple many-to-one references.

immediate: loaded as the parents are loaded, using a separate SELECT statement, or identity map fetch for simple many-to-one references.

joined: loaded eagerly in the same query as that of the parent, using a JOIN or LEFT OUTER JOIN. Whether the join is outer or not is determined by the relationship.innerjoin parameter.

subquery: loaded eagerly as the parents are loaded, using one additional SQL statement, which issues a JOIN to a subquery of the original statement, for each collection requested.

selectin: loaded eagerly as the parents are loaded, using one or more additional SQL statements, which issues a JOIN to the immediate parent object, specifying primary key identifiers using an IN clause.

noload: no loading should occur at any time. The related collection will remain empty. The noload strategy is not recommended for general use.

raise: lazy loading is disallowed; accessing the attribute, if its value were not already loaded via eager loading, will raise an InvalidRequestError. This strategy can be used when objects are to be detached from their attached Session after they are loaded.

raise_on_sql: lazy loading that emits SQL is disallowed.
Accessing the attribute, if its value were not already loaded via eager loading, will raise an InvalidRequestError, if the lazy load needs to emit SQL.

Example: defining models and their relationships

Define a person model with a one-to-many addresses relationship:

Explanations:
back_populates=’person’: allows objects on both sides of each relationship() to synchronize in-Python state changes
. It is generally specified on both sides.

lazy=’select’ : Specify is the default static loading of the relationship.
Here the select value means to load the relationship only if it is accessed/required.

cascade= »all, delete-orphan »: Specify how the persistence operations are cascaded on the relationship.
The default value is save-update, merge.
The all symbol is a synonym for save-update, merge, refresh-expire, expunge, delete.
And using it in conjunction with delete-orphan indicates that the child object should follow along with its parent in all cases, and be deleted once it is no longer associated with that parent.

from dataclasses import dataclass
from typing import List
 
from flask_example import db
from models.Address import Address
 
 
@dataclass
class Person(db.Model):
    id: int = db.Column(db.Integer, primary_key=True)
    firstname: str = db.Column(db.String)
    lastname = db.Column(db.String)
    addresses = db.relationship('Address',
                                back_populates='person',
                                lazy='select',
                                cascade="all, delete-orphan")
 
    # This constructor is not mandatory, but to have a check about parameter passed if we want to create 
    # manually a person instance, it is better to define it
    def __init__(self, firstname: str, lastname: str, addresses: List[str] = None):
        print(f'Person CONSTRUCTOR')
        self.lastname = lastname
        self.firstname = firstname
        if not addresses:
            addresses = []
        self.addresses = addresses
 
    def add_address(self, address: Address):
        self.addresses.append(address)
 
    def __repr__(self) -> str:
        return f'Person: id={self.id}, firstname={self.firstname}, lastname={self.lastname},' \
            # f'addresses={self.addresses}'

Define an address model with a many-to-one person relationship:

from dataclasses import dataclass
 
from flask_example import db
 
 
@dataclass
class Address(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    country_code: str = db.Column(db.String)
    zip_code = db.Column(db.String)
    street = db.Column(db.String)
    person = db.relationship('Person', back_populates='addresses')
 
    person_id = db.Column(db.Integer, db.ForeignKey('person.id'), nullable=False)
 
    def __init__(self, country_code: str, zip_code: str, street: str):
        print(f'Address CONSTRUCTOR')
        self.zip_code = zip_code
        self.country_code = country_code
        self.street = street
 
    def __repr__(self) -> str:
        return f'Address: id={self.id}, country_code={self.country_code}, zip_code={self.zip_code}, ' \
               f'street={self.street}'

Query examples

Function that insert or update a person

The merge() operation will delete all existing addresses rows and will recreate new addresses rows because we specified in the Person.addresses relationship mapping:
addresses = db.relationship('Address', ... cascade="all, delete-orphan")
Without this cascade configuration, new rows addresses are also created but the old rows will also stay in the database while not used any longer.

    def save(self, person: Person):
        if person.id:
            db.session.merge(person)
        else:
            db.session.add(person)
        db.session.commit()

Find person by id function that fetches the addresses relationship on demand

According to a boolean parameter, it loads only the entity requested or the entity plus its relationship.
We use the option joinedload to load eagerly and inside the same query the relationship.

    from typing import Optional
    from sqlalchemy.orm import joinedload
    from flask_example import db
    from models.Person import Person
 
    def find_by_id(self, person_id: int, fetch_address: bool) -> Optional[Person]:
        person: Person
        """:type: sqlalchemy.orm.Session"""
        query: db.Query = db.session.query(Person)
        if fetch_address:
            query = query.options(joinedload(Person.addresses))
        person: Person = query.get(person_id)
        return person

Find all persons functions with optional filters on person and its addresses relationship attributes

2 main goals:
– Retrieve all person objects.
– Retrieves person objects matching with person attributes and address attributes passed as parameters.
To perform the matching with address attribute, a join is performed.

from typing import List
import flask_sqlalchemy
import sqlalchemy
from flask_example import db
from models.Person import Person
 
    def find_all(self, lastname: str = None, country_code: str = None) -> List[Person]:
        """:type: sqlalchemy.orm.Session"""
        query: flask_sqlalchemy.query.Query = db.session.query(Person)
        if lastname:
            query = query.filter_by(lastname=lastname)
        if country_code:
            query = query.join(Person.addresses).filter_by(country_code=country_code)
        print(f'query={query}, type={type(query)}')
        persons: List[Person] = query.all()
        # SELECT person.id AS person_id, person.firstname AS person_firstname, person.lastname AS 
        # person_lastname
        # FROM person JOIN address ON person.id = address.person_id
        # WHERE person.lastname = %(lastname_1)s AND address.country_code = %(country_code_1)s
        # sqlalchemy.engine.Engine._exec_single_context  [cached since 9.263s ago] {'lastname_1': 'Jon',
        # 'country_code_1': 'FR'}
        return persons

Configuring Logging

Two ways

– Python’s standard logging module
– Use parameters create_engine.echo and create_engine.echo_pool on create_engine() which allow immediate logging to sys.stdout.

how to use it?

sqlalchemy.engine: SQL echoing.
Set to logging.INFO for SQL query output or logging.DEBUG for query + result set output.
Equivalent to echo=True and echo="debug" on create_engine.echo, respectively.

sqlalchemy.pool : connection pool logging.
Set to logging.INFO to log connection invalidation and recycle events or set to logging.DEBUG to additionally log all pool checkins and checkouts.
Equivalent to pool_echo=True and pool_echo="debug" on create_engine.echo_pool, respectively.

sqlalchemy.dialects : logging for SQL dialects.

sqlalchemy.orm : logging ORM functions.
Set to logging.INFO to log some top-level information on mapper configurations.

Declare a SQLAlchemy logger

import logging
 
logging.basicConfig()
logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)

marshmallow-sqlalchemy extension

It is a very helpful extension when we want to serialize/deserialize models to/from json.

Flask-SQLAlchemy