SQLAlchemy wrapper

The SQLAlchemy wrapper class is a light wrapper over regular SQLAlchemy, mainly to simplify the configuration.

A SQLAlchemy instance gives you access to the following things:

• db.engine: An engine created with the future=True argument
• A scoped session db.s and a db.Session class to manually create one, both extended with some useful active-record-like methods. (See “Working with the session”.)
• db.Model: A declarative base class
• db.create_all() and db.drop_all() methods to create and drop tables according to the models.
• db.test_transaction(): A helper for performant testing with a real database. (See “Testing with a real database”.)

Set up

The only required argument is the connection URI. You can give it directly:

from sqla_wrapper import SQLAlchemy

db = SQLAlchemy("postgresql://scott:tiger@localhost/test")

or as separated host, user, password, database name, etc. parameters, and SQLA-Wrapper will build the URI for you.

from sqla_wrapper import SQLAlchemy

db = SQLAlchemy(
    dialect="postgresql",
    user="scott",
    password="tiger",
    host="localhost",
    name="test",
)

After the setup, you will be interacting mostly directly with SQLAlchemy so I recommend reading the official SQLAlchemy tutorial if you haven’t done it yet.

Beyond the URI, the class also accepts an engine_options and a session_options dictionary to pass special options when creating the engine and/or the session.

Declaring models

A SQLAlchemy instance provides a db.Model class to be used as a declarative base class for your models. Follow the new type-based way to declare the table columns

from sqlalchemy.orm import Mapped, mapped_column
from myapp.models import db

class User(db.Model):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(128))

API

Create a SQLAlchemy connection

This class creates an engine, a base class for your models, and a scoped session.

The string form of the URL is dialect[+driver]://user:password@host/dbname[?key=value..], where dialect is a database name such as mysql, postgresql, etc., and driver the name of a DBAPI, such as psycopg2, pyodbc, etc.

Instead of the connection URL you can also specify dialect (plus optional driver), user, password, host, port, and database name as separate arguments.

Please review the Database URLs section of the SQLAlchemy documentation, for general guidelines in composing URL strings. In particular, special characters, such as those often part of passwords, must be URL-encoded to be properly parsed.

Example:

db = SQLAlchemy(database_uri)
# or SQLAlchemy(dialect=, name= [, user=] [, password=] [, host=] [, port=])

class Base(db.Model):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(sa.String(80), unique=True)
    deleted: Mapped[datetime] = mapped_column(sa.DateTime)

create_all(**kwargs)

Creates all the tables of the models registered so far.

Only tables that do not already exist are created. Existing tables are not modified.

drop_all(**kwargs)

Drop all the database tables.

Note that this is a destructive operation; data stored in the database will be deleted when this method is called.