Database functions

Simple thread-safe SQLAlchemy wrapper

after initialization, wrapper object contains additional params:

  • name equal to SQLAlchemy engine name
  • use_lastrowid show last row id be used (sqlite, mysql)
  • use_interval can interval columns be used (not available for sqlite, mysql)
  • parse_db_json should JSON column be parsed after selecting (sqlite, mysql)

__str__ and __repr__ methods are mapped to SQLAlchemy engine.

Extra mods required: sqlalchemy, msgpack (for KVStorage)

class pyaltt2.db.Database(dbconn=None, rq_func=None, **kwargs)

Database wrapper for SQLAlchemy

Parameters:
  • dbconn – database connection string (for SQLite - only file name is required)
  • rq_func – resource loader function (for query method)
  • kwargs – additional engine options (ignored for SQLite)
clone(**kwargs)

Clone database object

Extra kwargs (db, db_lock, g, rq_func) are assigned to object as-is

connect()

Get thread-safe db connection

create(q, *args, **kwargs)

Execute (usually INSERT) query with self.execute and return row id

row id must be in “id” field

execute(*args, _cr=False, **kwargs)

Execute SQL query

Parameters:
  • _cr – check result, raise LookupError if row count is zero
  • other – passed to SQLAlchemy connection as-is
get_engine()

Get DB engine object

list(*args, json_fields=[], **kwargs)

get self.execute result as list of dicts

Parameters:
  • json_fields – decode json fields if required
  • other – passed as-is
lookup(*args, json_fields=[], **kwargs)

Get single db row, use self.execute

Parameters:
  • json_fields – decode json fields if required
  • other – passed as-is
Returns:

single row as a dict
Raises:

LookupError – if nothing found
qcreate(q, *args, **kwargs)

Execute (usually INSERT) query with self.query and return row id

row id must be in “id” field

qlist(*args, json_fields=[], **kwargs)

get self.query result as list of dicts

Parameters:
  • json_fields – decode json fields if required
  • other – passed as-is
qlookup(*args, json_fields=[], **kwargs)

Get single db row, use self.query

Returns:single row as a dict
Raises:LookupError – if nothing found
query(q, qargs=[], qkwargs={}, _create=False, *args, **kwargs)

Execute SQL query by resource

Parameters:
  • q – resource name
  • qkwargs (qargs,) – format query with args/kwargs
  • other – passed as-is

Requires rq_func

class pyaltt2.db.KVStorage(db, table_name='kv')

Simple key-value database storage

Parameters:
  • db – pyaltt2.db.Database
  • table_name – storage table name (default: kv)
cleanup()

Deletes expired objects

delete(key)

Delete object in key-value storage

Parameters:key – object key
Raises:LookupError – object not found
get(key, delete=False)

Get object from key-value storage

Parameters:
  • key – object key
  • delete – delete object after getting
Raises:

LookupError – object not found
put(key=None, value=None, expires=None, override=True)

Put object to key-value storage

If no key specified, random 64-char key is generated

Parameters:
  • key – string key (1-255 chars)
  • value – value to put
  • expires – expiration either in seconds or datetime.timedelta
  • override – replace existing object
Returns:

object key
pyaltt2.db.format_condition(f, kw=None, fields=None, cond=None)
Parameters:
  • f – condition filter (dict name/value)
  • kw – kwargs query dict
  • fields – allowed fields
  • cond – initial condition
Returns:

tuple cond (string), kw (dict). String can be safely used in SQL query, dict should be set as query kwargs