Source

Default


def Default(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Initialize self. See help(type(self)) for accurate signature.

`Database` and `DBTable`

We create a Database class and a DBTable class (which is returned by Database.create), using sqlalchemy v2. These classes will allow us to directly work with dataclasses such as these:

conn_str = _db_str(":memory:"); conn_str

'sqlite:///:memory:'

source

Database


def Database(
    conn_str, engine_kws:NoneType=None
):

A connection to a SQLAlchemy database

db = Database(conn_str); db

Database(sqlite:///:memory:)

# create a test table
db.execute(sa.text('create table test (id integer primary key, name text, age integer)'))
db.meta.reflect(bind=db.engine)

db.meta.tables

FacadeDict({'test': Table('test', MetaData(), Column('id', INTEGER(), table=<test>, primary_key=True), Column('name', TEXT(), table=<test>), Column('age', INTEGER(), table=<test>), schema=None)})

source

Database.q


def q(
    sql:str, params:VAR_KEYWORD
):

Query database with raw SQL and optional parameters. Returns list of dicts.

# add some data to it
db.q('insert into test (name, age) values (:name, :age)', name='Alice', age=30)
db.q('insert into test (name, age) values (:name, :age)', name='Bob', age=25)

[]

db.q('select * from test')

[{'id': 1, 'name': 'Alice', 'age': 30}, {'id': 2, 'name': 'Bob', 'age': 25}]

# update Alice's age to 32
db.q('update test set age = :age where name = :name', age=32, name='Alice')

[]

db.q('select * from test')

[{'id': 1, 'name': 'Alice', 'age': 32}, {'id': 2, 'name': 'Bob', 'age': 25}]

source

DBTable


def DBTable(
    table:Table, db:Database, cls, _exists:NoneType=None
):

A connection to a SQLAlchemy table, created if needed

# create a DBTable with the Test table
tbl = DBTable(db.meta.tables['test'], db, None)
tbl

<Table test (id, name, age)>

source

DBTable.t


def t(
    
):

tbl.t

(Table('test', MetaData(), Column('id', INTEGER(), table=<test>, primary_key=True), Column('name', TEXT(), table=<test>), Column('age', INTEGER(), table=<test>), schema=None),
 <sqlalchemy.sql.base.ReadOnlyColumnCollection>)

source

DBTable.pks


def pks(
    
):

tbl.pks

(Column('id', INTEGER(), table=<test>, primary_key=True),)

source

DBTable.schema


def schema(
    
):

print(tbl.schema)

CREATE TABLE test (
    id INTEGER, 
    name TEXT, 
    age INTEGER, 
    PRIMARY KEY (id)
)

source

Database.table


def table(
    nm:str, cls:NoneType=None
):

db.table('test')

<Table test (id, name, age)>

source

Database.getitem


def __getitem__(
    nm:str
):

db['test']

<Table test (id, name, age)>

source

database


def database(
    path, wal:bool=True, kwargs:VAR_KEYWORD
)->Any:

Create a Database from a path or connection string

db = database("sqlite:///:memory:"); db

Database(sqlite:///:memory:)

source

create_mod


def create_mod(
    db, mod_fn, with_views:bool=False, store:bool=True, suf:str=''
):

Create module for dataclasses for db

source

all_dcs


def all_dcs(
    db, with_views:bool=False, store:bool=True, suf:str=''
):

dataclasses for all objects in db

source

Database.get_tables


def get_tables(
    glb
):

Add objects for all table objects to namespace glb

source

Database.set_classes


def set_classes(
    glb
):

Add set all table dataclasses using types in namespace glb

source

Database.link_dcs


def link_dcs(
    mod
):

Set the internal dataclass type links for tables using mod (created via create_mod)

source

DBTable.lookup


def lookup(
    lookup_values:Dict, extra_values:Optional=None, pk:str | None='id', kwargs:VAR_KEYWORD
):

source

DBTable.upsert


def upsert(
    record:Any=None, pk:Default=<__main__.Default object at 0x7fe1bb4866f0>,
    foreign_keys:Default=<__main__.Default object at 0x7fe1bb4866f0>,
    column_order:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    not_null:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    defaults:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    hash_id:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    hash_id_columns:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    alter:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    extracts:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    conversions:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    columns:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    strict:Union=<__main__.Default object at 0x7fe1bb4866f0>, kwargs:VAR_KEYWORD
)->Any:

CRUD Operations

source

get_typ


def get_typ(
    t
):

Get the underlying type.

get_typ(int|None), get_typ(list[str])

(int, str)

_sa_type(int)

sqlalchemy.sql.sqltypes.Integer

_column('age', int)

Column('age', Integer(), table=None)

source

Database.create


def create(
    cls:type, pk:str='id', name:str | None=None, foreign_keys:NoneType=None, defaults:NoneType=None,
    column_order:NoneType=None, not_null:NoneType=None, hash_id:NoneType=None, hash_id_columns:NoneType=None,
    extracts:NoneType=None, if_not_exists:bool=False, replace:bool=False, ignore:bool=True, transform:bool=False,
    strict:bool=False, constraints:NoneType=None
):

Get a table object, creating in DB if needed

class User: name:str; pwd:str
class Todo: title:str; name:str; id:int; done:bool=False; details:str=''
class Student: id:int; grad_year:int; name:str

users = db.create(User, pk='name')
todos = db.create(Todo, pk='id')
students = db.create(Student, pk=('id', 'grad_year'))

source

Database.view_names


def view_names(
    
):

source

Database.table_names


def table_names(
    
):

db.table_names()

['student', 'todo', 'user']

source

Database.schema


def schema(
    
):

Show all tables and columns

print(db.schema())

Table: student
  * id: INTEGER
  * grad_year: INTEGER
  - name: VARCHAR
Table: todo
  - title: VARCHAR
  - name: VARCHAR
  * id: INTEGER
  - done: BOOLEAN
  - details: VARCHAR
Table: user
  * name: VARCHAR
  - pwd: VARCHAR

source

Database.t


def t(
    
):

By returning a _TablesGetter we get a repr and auto-complete that shows all tables in the DB.

db.t

student, todo, user

source

DBTable.c


def c(
    
):

Similarly, we return _ColsGetter for a table’s columns

users.c

name, pwd

source

DBTable.create


def create(
    columns:Dict=None, pk:Any=None, foreign_keys:NoneType=None, column_order:Optional=None, not_null:Optional=None,
    defaults:Optional=None, hash_id:str | None=None, hash_id_columns:Optional=None, extracts:Union=None,
    if_not_exists:bool=False, replace:bool=False, ignore:bool=False, transform:bool=False, strict:bool=False,
    kwargs:VAR_KEYWORD
):

Create table from column definitions passed as kwargs or columns dict

source

DBTable.exists


def exists(
    
):

Check if this table exists in the DB

users.exists()

True

u0 = User('jph','foo')
u1 = User('rlt','bar')
t0 = Todo('do it', 'jph')
t1 = Todo('get it done', 'rlt')

source

DBTable.insert


def insert(
    record:Dict=None, pk:Default=<__main__.Default object at 0x7fe1bb4866f0>,
    foreign_keys:Default=<__main__.Default object at 0x7fe1bb4866f0>,
    column_order:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    not_null:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    defaults:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    hash_id:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    hash_id_columns:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    alter:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    ignore:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    replace:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    extracts:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    conversions:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    columns:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    strict:Union=<__main__.Default object at 0x7fe1bb4866f0>, kwargs:VAR_KEYWORD
)->Any:

Insert an object into this table, and return it

t = todos.insert(t0)
assert t.id
t

Todo(title='do it', name='jph', id=1, done=False, details='')

u = users.insert(u0)
assert u.name=='jph'
users.insert(u1)
u

User(name='jph', pwd='foo')

todos.insert(t1)

Todo(title='get it done', name='rlt', id=2, done=False, details='')

source

DBTable.insert_all


def insert_all(
    records:Optional=None, pk:Default=<__main__.Default object at 0x7fe1bb4866f0>,
    foreign_keys:Default=<__main__.Default object at 0x7fe1bb4866f0>,
    column_order:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    not_null:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    defaults:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    batch_size:Default=<__main__.Default object at 0x7fe1bb4866f0>,
    hash_id:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    hash_id_columns:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    alter:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    ignore:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    replace:Union=<__main__.Default object at 0x7fe1bb4866f0>, truncate:bool=False,
    extracts:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    conversions:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    columns:Union=<__main__.Default object at 0x7fe1bb4866f0>,
    strict:Union=<__main__.Default object at 0x7fe1bb4866f0>, upsert:bool=False, analyze:bool=False,
    xtra:dict | None=None, kwargs:VAR_KEYWORD
)->DBTable:

todos.insert_all([t0,t1])

<Table todo (title, name, id, done, details)>

source

DBTable.count_where


def count_where(
    where:Optional=None, where_args:Union=None, kw:VAR_KEYWORD
)->int:

source

DBTable.count


def count(
    
):

source

DBTable.len


def __len__(
    
):

todos.count, len(todos)

(4, 4)

source

DBTable.rows_where


def rows_where(
    where:Optional=None, where_args:Union=None, order_by:Optional=None, select:str='*', limit:Optional=None,
    offset:Optional=None, xtra:dict | None=None, kw:VAR_KEYWORD
)->Generator:

source

DBTable.pks_and_rows_where


def pks_and_rows_where(
    where:Optional=None, where_args:Union=None, order_by:Optional=None, select:str='*', limit:Optional=None,
    offset:Optional=None, xtra:dict | None=None, kw:VAR_KEYWORD
)->Generator:

source

DBTable.rows


def rows(
    
):

source

DBTable.call


def __call__(
    where:str | None=None, # SQL where fragment to use, for example `id > ?`
    where_args:Union=None, # Parameters to use with `where`; iterable for `id>?`, or dict for `id>:id`
    order_by:str | None=None, # Column or fragment of SQL to order by
    limit:int | None=None, # Number of rows to limit to
    offset:int | None=None, # SQL offset
    select:str='*', # Comma-separated list of columns to select
    with_pk:bool=False, # Return tuple of (pk,row)?
    as_cls:bool=True, # Convert returned dict to stored dataclass?
    xtra:dict | None=None, # Extra constraints
    kw:VAR_KEYWORD
):

Result of select query on the table. Returns list of returned objects.

assert users()==[u0,u1]
users()

[User(name='jph', pwd='foo'), User(name='rlt', pwd='bar')]

r = users(where="pwd LIKE :pwd", pwd="b%")
assert r==[u1]
r

[User(name='rlt', pwd='bar')]

users.xtra(name='rlt')
users(order_by='name')

[User(name='rlt', pwd='bar')]

users(where='name="rlt"')

[User(name='rlt', pwd='bar')]

users(where='name="jph"')

[]

assert len(todos())==4
todos()

[Todo(title='do it', name='jph', id=1, done=False, details=''),
 Todo(title='get it done', name='rlt', id=2, done=False, details=''),
 Todo(title='do it', name='jph', id=3, done=False, details=''),
 Todo(title='get it done', name='rlt', id=4, done=False, details='')]

source

Database.create_view


def create_view(
    name:str, sql:str, ignore:bool=False, replace:bool=False
):

Create a view with the given name and SQL query

# Create a view showing only incomplete todos
db.create_view('pending_todos', 'SELECT * FROM todo WHERE done = 0')

source

Database.v


def v(
    
):

db.v.pending_todos()

[{'title': 'do it', 'name': 'jph', 'id': 1, 'done': False, 'details': ''},
 {'title': 'get it done',
  'name': 'rlt',
  'id': 2,
  'done': False,
  'details': ''},
 {'title': 'do it', 'name': 'jph', 'id': 3, 'done': False, 'details': ''},
 {'title': 'get it done',
  'name': 'rlt',
  'id': 4,
  'done': False,
  'details': ''}]

source

MissingPrimaryKey


def MissingPrimaryKey(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Common base class for all non-exit exceptions.

source

NotFoundError


def NotFoundError(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Common base class for all non-exit exceptions.

source

DBTable.getitem


def __getitem__(
    key
):

source

DBTable.get


def get(
    pk_values, as_cls:bool=True, xtra:dict | None=None, default:Any=UNSET
)->Any:

Get item with PK key

users.xtra(name='jph')
assert users['jph']==u0
users['jph']

User(name='jph', pwd='foo')

users.xtra(name='rlt')
test_fail(lambda: users['jph']==u0)

source

DBTable.selectone


def selectone(
    where:str | None=None, where_args:Union=None, select:str='*', as_cls:bool=True, xtra:dict | None=None,
    kwargs:VAR_KEYWORD
):

users.xtra()  # clear xtra
test_eq(users.selectone('name=?', ['jph']).name, 'jph')

source

DBTable.update


def update(
    updates:dict | None=None, pk_values:list | tuple | str | int | float | None=None, alter:bool=False,
    conversions:dict | None=None, xtra:dict | None=None, kwargs:VAR_KEYWORD
)->Any:

users.xtra(name='jph')
u.pwd = 'new'
users.update(u)
users.xtra()
users()

[User(name='jph', pwd='new'), User(name='rlt', pwd='bar')]

users.xtra(name='rlt')
u.pwd = 'foo'
users.update(u)
users.xtra()
test_eq(users['jph'].pwd, 'new')

source

DBTable.update_where


def update_where(
    updates:dict, where:str | None=None, where_args:Union=None, xtra:dict | None=None, kw:VAR_KEYWORD
)->list:

Update rows matching where with updates. Returns updated rows.

source

DBTable.update_where


def update_where(
    updates:dict, where:str | None=None, where_args:Union=None, xtra:dict | None=None, kw:VAR_KEYWORD
)->list:

Update rows matching where with updates. Returns updated rows.

todos.update_where({'done': True}, where='name = :name', name='jph')

[Todo(title='do it', name='jph', id=1, done=True, details=''),
 Todo(title='do it', name='jph', id=3, done=True, details='')]

source

DBTable.delete


def delete(
    key
):

Delete item with PK key and return the deleted object

assert users.delete('jph')
test_fail(lambda: users['jph'])

source

DBTable.delete_where


def delete_where(
    where:Optional=None, where_args:Union=None, xtra:dict | None=None, kw:VAR_KEYWORD
):

todos.delete_where("name = ?", ["jph"])

[Todo(title='do it', name='jph', id=1, done=True, details=''),
 Todo(title='do it', name='jph', id=3, done=True, details='')]

todos('name=?', ['jph'])

[]

source

DBTable.contains


def __contains__(
    pk_values:Union
)->bool: # A single value, or a tuple of values for tables that have a compound primary key

Is the item with the specified primary key value in this table?

Demonstration with single field primary key:

assert not 'jph' in users
assert 'rlt' in users

For compound primary keys, lets whether a student is in the students table or not.

students.insert(Student(1, 2021, 'jph'))

Student(id=1, grad_year=2021, name='jph')

assert (1,2021) in students
assert (1,2030) not in students

source

DBTable.drop


def drop(
    ignore:bool=False
):

Drop this table from the database

students.drop()
assert 'student' not in db.t

Migrations

db.t

todo, user

source

Database.migrate


def migrate(
    mdir
):

db.version = 0

db.migrate(mdir)
test_eq(db.version, 1)
assert 'priority' in db.t.todo.c
assert 'due_date' in db.t.todo.c

Applied migration 1: 1-add_priority_and_due_date_to_todo.sql

db.t.todo()

[Todo(title='get it done', name='rlt', id=2, done=False, details='', priority=0, due_date=None),
 Todo(title='get it done', name='rlt', id=4, done=False, details='', priority=0, due_date=None)]

SQLAlchemy helpers

source

ReadOnlyColumnCollection.dir


def __dir__(
    
):

Default dir() implementation.

source

MetaData.dir


def __dir__(
    
):

Default dir() implementation.

dbm = db.meta

' '.join(dbm.tables)

'_meta todo user'

t = dbm.todo

list(t.c)

[Column('title', VARCHAR(), table=<todo>),
 Column('name', VARCHAR(), table=<todo>),
 Column('id', INTEGER(), table=<todo>, primary_key=True),
 Column('done', BOOLEAN(), table=<todo>),
 Column('details', VARCHAR(), table=<todo>),
 Column('priority', INTEGER(), table=<todo>, server_default=DefaultClause(<sqlalchemy.sql.elements.TextClause object>, for_update=False)),
 Column('due_date', TEXT(), table=<todo>)]

from sqlalchemy.exc import ResourceClosedError

source

MetaData.sql


def sql(
    statement, args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Execute statement string and return DataFrame of results (if any)

source

Connection.sql


def sql(
    statement, nm:str='Row', args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Execute statement string and return results (if any)

source

CursorResult.tuples


def tuples(
    nm:str='Row'
):

Get all results as named tuples

# dbm.sql('delete from todo')
# db.conn.commit()

rs = dbm.sql('select * from user')
rs[0]

Row(name='rlt', pwd='foo')

source

Table.get


def get(
    where:NoneType=None, limit:NoneType=None
):

Select from table, optionally limited by where and limit clauses

t.get(t.c.title.startswith('d'), limit=5)

[]

This is the query that will run behind the scenes:

print(t.select().where(t.c.title.startswith('d')).limit(5))

SELECT todo.title, todo.name, todo.id, todo.done, todo.details, todo.priority, todo.due_date 
FROM todo 
WHERE (todo.title LIKE :title_1 || '%')
 LIMIT :param_1

source

MetaData.close


def close(
    
):

Close the connection

dbm.close()

Default

Database and DBTable

Database

Database.q

DBTable

DBTable.t

DBTable.pks

DBTable.schema

Database.table

Database.__getitem__

database

create_mod

all_dcs

Database.get_tables

Database.set_classes

Database.link_dcs

DBTable.lookup

DBTable.upsert

CRUD Operations

get_typ

Database.create

Database.view_names

Database.table_names

Database.schema

Database.t

DBTable.c

DBTable.create

DBTable.exists

DBTable.insert

DBTable.insert_all

DBTable.count_where

DBTable.count

DBTable.__len__

DBTable.rows_where

DBTable.pks_and_rows_where

DBTable.rows

DBTable.__call__

Database.create_view

Database.v

MissingPrimaryKey

NotFoundError

DBTable.__getitem__

DBTable.get

DBTable.selectone

DBTable.update

DBTable.update_where

DBTable.update_where

DBTable.delete

DBTable.delete_where

DBTable.__contains__

DBTable.drop

Migrations

Meta

Database.migrate

SQLAlchemy helpers

ReadOnlyColumnCollection.__dir__

MetaData.__dir__

MetaData.sql

Connection.sql

CursorResult.tuples

Table.get

MetaData.close

`Database` and `DBTable`

Database.getitem

DBTable.len

DBTable.call

DBTable.getitem

DBTable.contains

ReadOnlyColumnCollection.dir

MetaData.dir