Python sqlalchemy

Documentation

official docs	https://docs.sqlalchemy.org/en/13/
column apidoc	https://docs.sqlalchemy.org/en/13/core/metadata.html?highlight=sqlalchemy%20column#sqlalchemy.schema.Column
column datatypes	https://docs.sqlalchemy.org/en/13/core/type_basics.html

QueuePool without ORM

import sqlalchemy.pool
import pymysql

def new_connection():
    conn = pymysql.connect( user='root', host='127.0.0.1', password='password', db='test_database', charset='utf-8')
    return conn

mypool = sqlalchemy.pool.QueuePool( new_connection, max_overflow=10, pool_size=5)

## returns connections from pool
conn = mypool.connect()
conn = mypool.connect()
conn = mypool.connect()
...

ORM

Mapping to Database

It is likely that all of this will exist in the same module, or at least the same package. This produces all required classes to connect to a database.

Create classes to represent database

from sqlalchemy.ext import declarative
import sqlalchemy as sql

class User(Base):
    __tablename__ = 'user'

    id = sql.Column(
        sql.Integer, nullable=False, unique=True, autoincrement=True, primary_key=True,
        comment='User Id',
    )

    name = sql.Column(
        sql.Text, nullable=False, unique=True,
        comment='The name of the user',
    )

Create Connection Engine

def engine():
    return sqlite.create_engine('sqlite+pysqlite:///{}'.format('/path/to/file.sqlite'))

Create Session class

Session = orm.sessionmaker(bind=engine())

Creating/Updating Database Schema (for all Base subclasses found in current module)

Base.metadata.create_all(engine)

Session

All of the ORM's functionality is built upon the Session object. A session is an object like a transaction. It can be committed or rolled back. Sessions also support transactions.

Insert

session = Session()
session.add(YourTable(col='value'))
session.add_all([
    YourTable(col='value'),
    YourTable(col='value'),
])

Query

# non python comparison
query.filter_by(Table.col='value')

# python comparison
query.filter(Table.col == 'value')

# chain together filters. Nothing evaluated until all(), count(), first(), ...
query.filter(YourTable.first == 'John').filter(YourTable.last == 'Doe')

query = session.query(Table)

query.filter(Table.col=='value')                                  # equals
query.filter(Table.col!='value')                                  # not equals
query.filter(Table.col.like('value'))                             # like
query.filter(Table.col.ilike('value'))                            # case-insensitive like
query.filter(Table.col.in_(['a','b','c']))                        # in 
query.filter(~Table.col.in_(['a','b','c']))                       # not in
query.filter(Table.col.is_(None))                                 # is
query.filter(Table.col.isnot(None))                               # is not
query.filter(sqlalchemy.or_(Table.col == 'a', Table.col == 'b'))  # or

Get Query Results

session.query(YourTable).order_by(YourTable.id)     # sort
session.query(YourTable).all()                      # all results
session.query(YourTable).count()                    # number of results
session.query(YourTable).first()                    # first result
session.query(YourTable).one()                      # one result (error on no results)
session.query(YourTable).one_or_none()              # one result or none (no error on no results)