Source

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:

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

source

Database

 Database (conn_str)

A connection to a SQLAlchemy database

db = Database("sqlite:///:memory:")
# db = Database("postgresql://")

source

DBTable

 DBTable (table:sqlalchemy.sql.schema.Table, database:__main__.Database,
          cls)

A connection to a SQLAlchemy table, created if needed

source

Database.create

 Database.create (cls:type, pk='id', name:str|None=None)

Get a table object, creating in DB if needed

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

source

Database.schema

 Database.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

DBTable.exists

 DBTable.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

 DBTable.insert (obj)

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.__call__

 DBTable.__call__ (where:str|None=None,
                   where_args:Union[Iterable,dict,NoneType]=None,
                   order_by:str|None=None, limit:int|None=None,
                   offset:int|None=None, select:str='*', **kw)

Result of select query on the table

Type Default Details
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
kw
Returns list 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())==2
todos()
[Todo(title='do it', name='jph', id=1, done=False, details=''),
 Todo(title='get it done', name='rlt', id=2, done=False, details='')]

source

NotFoundError

Common base class for all non-exit exceptions.

source

DBTable.__getitem__

 DBTable.__getitem__ (key)

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.update

 DBTable.update (obj=None, **kw)
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.delete

 DBTable.delete (key)

Delete item with PK key and return count deleted

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

source

DBTable.__contains__

 DBTable.__contains__ (pk_values:Union[list,tuple,str,int])

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

Type Details
pk_values Union A single value, or a tuple of values for tables that have a compound primary key
Returns bool

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

SQLAlchemy helpers

source

ReadOnlyColumnCollection.__dir__

 ReadOnlyColumnCollection.__dir__ ()

Default dir() implementation.

source

MetaData.__dir__

 MetaData.__dir__ ()

Default dir() implementation.

dbm = db.meta
' '.join(dbm.tables)
'user todo student'
t = dbm.todo
list(t.c)
[Column('title', String(), table=<todo>),
 Column('name', String(), table=<todo>),
 Column('id', Integer(), table=<todo>, primary_key=True, nullable=False),
 Column('done', Boolean(), table=<todo>),
 Column('details', String(), table=<todo>)]
from sqlalchemy.exc import ResourceClosedError

source

MetaData.sql

 MetaData.sql (statement, *args, **kwargs)

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

source

Connection.sql

 Connection.sql (statement, nm='Row', *args, **kwargs)

Execute statement string and return results (if any)

source

CursorResult.tuples

 CursorResult.tuples (nm='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

 Table.get (where=None, limit=None)

Select from table, optionally limited by where and limit clauses

t.get(t.c.title.startswith('d'), limit=5)
[Row(title='do it', name='jph', id=1, done=False, details='')]

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 
FROM todo 
WHERE (todo.title LIKE :title_1 || '%')
 LIMIT :param_1

source

MetaData.close

 MetaData.close ()

Close the connection

dbm.close()