cs-sqlalchemy-utils 20260531

Assorted utility functions to support working with SQLAlchemy.

Assorted utility functions to support working with SQLAlchemy.

Latest release 20260531:

• Expose the db URL normalisation stuff as methods.
• ORM.startup_shutdown: do not obtain the db lock, we will now do this per orchestrated_session().
• New ORM.dbshell() method to run an interactive db prompt.

Short summary:

• auto_session: A decorator to run a function in a session if one is not presupplied. The function function runs within a transaction, nested if the session already exists.
• BasicTableMixin: Useful methods for most tables.
• find_json_field: Descend a JSONable Python object column_value to field_name. Return column_value (possibly infilled), final_field, final_field_name.
• get_json_field: Return the value of field_name from column_value or a defaault if the field is not present.
• HasIdMixin: Include an "id" Column as the primary key.
• json_column: Class decorator to declare a virtual column name on a table where the value resides inside a JSON column of the table.
• log_level: Temporarily set the level of the default SQLAlchemy logger to level. Yields the logger.
• ORM: A convenience base class for an ORM class.
• orm_auto_session: A decorator to run a method in a session derived from self.orm if a session is not presupplied. This is intended to assist classes with a .orm attribute.
• proxy_on_demand_field: A decorator to provide a field value on demand via a function field_func(self,db_row,session=session).
• RelationProxy: Construct a proxy for a row from a relation.
• set_json_field: Set a new value for field_name of column_value. Return the new column_value.
• SQLAState: Thread local state for SQLAlchemy ORM and session.
• state: Thread local state for SQLAlchemy ORM and session.
• using_session: A context manager to prepare an SQLAlchemy session for use by a suite.
• with_orm: Call function with the supplied orm in the shared state.
• with_session: Call function(*a,session=session,**kw), creating a session if required. The function function runs within a transaction, nested if the session already exists. If a new session is created it is set as the default session in the shared state.

Module contents:

• auto_session(*da, **dkw): A decorator to run a function in a session if one is not presupplied. The function function runs within a transaction, nested if the session already exists.

  See with_session for details.

• class BasicTableMixin: Useful methods for most tables.

BasicTableMixin.__getitem__(index, *, id_column=None, session): Index the table by its id_column column, default 'id'.

BasicTableMixin.by_id(index, *, id_column=None, session): Index the table by its id_column column, default 'id'.

BasicTableMixin.lookup(*, session, **criteria): Return an iterable Query of row entities matching criteria.

BasicTableMixin.lookup1(*, session, **criteria): Return the row entity matching criteria, or None if no match.

• find_json_field(column_value, field_name, *, infill=False): Descend a JSONable Python object column_value to field_name. Return column_value (possibly infilled), final_field, final_field_name.

  This supports database row columns which are JSON columns.

  Parameters:

  • column_value: the original value of the column
  • field_name: the field within the column to locate
  • infill: optional keyword parameter, default False. If true, column_value and its innards will be filled in as dicts to allow deferencing the field_name.

  The field_name is a str consisting of a period ('.') separated sequence of field parts. Each field part becomes a key to index the column mapping. These keys are split into the leading field parts and the final field part, which is returned as final_field_name above.

  The final_field return value above is the mapping within which final_field_value may lie and where final_field_value may be set. Note: it may not be present.

  If a leading key is missing and infill is true the corresponding part of the column_value is set to an empty dictionary in order to allow deferencing the leading key. This includes the case when column_value itself is None, which is why the column_value is part of the return.

  If a leading key is missing and infill is false this function will raise a KeyError for the portion of the field_name which failed.

  Examples:

  >>> find_json_field({'a':{'b':{}}}, 'a.b')
  ({'a': {'b': {}}}, {'b': {}}, 'b')
  >>> find_json_field({'a':{}}, 'a.b')
  ({'a': {}}, {}, 'b')
  >>> find_json_field({'a':{'b':{}}}, 'a.b.c.d')
  Traceback (most recent call last):
      ...
  KeyError: 'a.b.c'
  >>> find_json_field({'a':{'b':{}}}, 'a.b.c.d', infill=True)
  ({'a': {'b': {'c': {}}}}, {}, 'd')
  >>> find_json_field(None, 'a.b.c.d')
  Traceback (most recent call last):
      ...
  KeyError: 'a'
  >>> find_json_field(None, 'a.b.c.d', infill=True)
  ({'a': {'b': {'c': {}}}}, {}, 'd')
  

• get_json_field(column_value, field_name, *, default=None): Return the value of field_name from column_value or a defaault if the field is not present.

  Parameters:

  • column_value: the original value of the column
  • field_name: the field within the column to locate
  • default: default value to return if the field is not present, default: None

  Examples:

  >>> get_json_field({'a': 1}, 'a')
  1
  >>> get_json_field({'b': 1}, 'a')
  >>> get_json_field({'a': {}}, 'a.b')
  >>> get_json_field({'a': {'b': 2}}, 'a.b')
  2
  

• class HasIdMixin: Include an "id" Column as the primary key.

• json_column(*da, **dkw): Class decorator to declare a virtual column name on a table where the value resides inside a JSON column of the table.

  Parameters:

  • cls: the class to annotate
  • attr: the virtual column name to present as a row attribute
  • json_field_name: the field within the JSON column used to store this value, default the same as attr
  • json_column_name: the name of the associated JSON column, default 'info'
  • default: the default value returned by the getter if the field is not present, default None

  Example use:

  Base = declarative_base()
  ...
  @json_column('virtual_name', 'json.field.name')
  class TableClass(Base):
    ...
  

  This annotates the class with a .virtual_name property which can be accessed or set, accessing or modifying the associated JSON column (in this instance, the column info, accessing info['json']['field']['name']).

• log_level(func, a, kw, level=None): Temporarily set the level of the default SQLAlchemy logger to level. Yields the logger.

  NOTE: this is not MT safe - competing Threads can mix log levels up.

• class ORM(cs.resources.MultiOpenMixin): A convenience base class for an ORM class.

  This defines a .Base attribute which is a new DeclarativeBase and provides various Session related convenience methods. It is also a MultiOpenMixin subclass supporting nested open/close sequences and use as a context manager.

ORM.__init__(self, db_url, serial_sessions=None): Initialise the ORM.

The db_url is any database URL of the form dbtype://.... It may also be an absolute filesystem path, starting with / and ending in .sqlite. This is transformed into an sqlite:///fspath URL.

If serial_sessions is true (default True for SQLite, False otherwise) then allocate a lock to serialise session allocation. This might be chosen with SQL backends which do not support concurrent sessions such as SQLite.

In the case of SQLite there's a small inbuilt timeout in an attempt to serialise transactions but it is possible to exceed it easily and recovery is usually infeasible. Instead we use the serial_sessions option to obtain a mutex before allocating a session.

ORM.dbshell(self): Run an interactive database prompt.

ORM.declare_schema(self): Declare the database schema / ORM mapping. This just defines the relation types etc. It does not act on the database itself. It is called automatically at the end of __init__.

Example:

def declare_schema(self):
  """ Define the database schema / ORM mapping.
  """
  orm = self
  Base = self.Base
  class Entities(
  ........
  self.entities = Entities

After this, methods can access the example Entities relation as self.entites.

ORM.default_session: The current per-Thread session.

ORM.engine: SQLAlchemy engine, made on demand.

ORM.fspath_for_db_url(db_url): Return the filesystem path for a db_url, or None. This extracts the path from sqlite:/// URLs.

ORM.norm_db_url(db_url: str) -> str: Promote a /path.sqlite filesystem path to an sqlite:/// URL. Return other strings unchanged.

ORM.orchestrated_session(self): A context manager to orchestrate a new session for this Thread, honouring self.serial_session and yielding the new Session.

ORM.serialif(self): A context manager to serialise sessions if self.serial_sessions.

ORM.startup_shutdown(self): Default startup/shutdown context manager. This updates the database schema on first use.

• orm_auto_session(*da, **dkw): A decorator to run a method in a session derived from self.orm if a session is not presupplied. This is intended to assist classes with a .orm attribute.

  See with_session for details.

• proxy_on_demand_field(*da, **dkw): A decorator to provide a field value on demand via a function field_func(self,db_row,session=session).

  Example:

  @property
  @proxy_on_demand_field
  def formats(self,db_row,*,session):
      """ A mapping of Calibre format keys to format paths
          computed on demand.
      """
      return {
          fmt.format:
          joinpath(db_row.path, f'{fmt.name}.{fmt.format.lower()}')
          for fmt in db_row.formats
      }
  

• RelationProxy(relation, columns: Union[str, Tuple[str], List[str]], *, id_column: Optional[str] = None, orm=None, missing=None): Construct a proxy for a row from a relation.

  Parameters:

  • relation: an ORM relation for which this will be a proxy
  • columns: a list of the column names to cache, or a space separated string of the column names
  • id_column: options primary key column name, default from BasicTableMixin.DEFAULT_ID_COLUMN: 'id'
  • orm: the ORM, default from relation.orm
  • missing: an optional function to produce a default value for a field if there is no matching record in the relation; if None (the default) access to a field raises KeyError or AttributeError as appropriate

  This is something of a workaround for applications which dip briefly into the database to obtain information instead of doing single long running transactions or sessions. Instead of keeping the row instance around, which might want to load related data on demand after its source session is expired, we keep a proxy for the row with cached values and refetch the row at need if further information is required.

  Typical use is to construct this proxy class as part of the __init__ of a larger class which accesses the database as part of its operation. The example below is based on cs.ebooks.calibre.CalibreTree:

  def __init__(self, calibrepath):
    super().__init__(calibrepath)
    # define the proxy classes
    class CalibreBook(RelationProxy(self.db.books, [
        'author',
        'title',
    ])):
      """ A reference to a book in a Calibre library.
      """
      @typechecked
      def __init__(self, tree: CalibreTree, dbid: int, db_book=None):
        self.tree = tree
        self.dbid = dbid
      ... various other CalibreBook methods ...
    self.CalibreBook = CalibreBook
  
  def __getitem__(self, dbid):
    return self.CalibreBook(self, dbid, db_book=db_book)
  

• set_json_field(column_value, field_name, value, *, infill=False): Set a new value for field_name of column_value. Return the new column_value.

  Parameters:

  • column_value: the original value of the column
  • field_name: the field within the column to locate
  • value: the value to store as field_name
  • infill: optional keyword parameter, default False. If true, column_value and its innards will be filled in as dicts to allow deferencing the field_name.

  As with find_json_field, a true infill may modify column_value to provide field_name which is why this function returns the new column_value.

  Examples:

  >>> set_json_field({'a': 2}, 'a', 3)
  {'a': 3}
  >>> set_json_field({'a': 2, 'b': {'c': 5}}, 'b.c', 4)
  {'a': 2, 'b': {'c': 4}}
  >>> set_json_field({'a': 2}, 'b.c', 4)
  Traceback (most recent call last):
      ...
  KeyError: 'b'
  >>> set_json_field({'a': 2}, 'b.c', 4, infill=True)
  {'a': 2, 'b': {'c': 4}}
  >>> set_json_field(None, 'b.c', 4, infill=True)
  {'b': {'c': 4}}
  

• class SQLAState(cs.threads.ThreadState): Thread local state for SQLAlchemy ORM and session.

SQLAState.auto_session(self, *, orm=None): Context manager to use the current session if not None, otherwise to make one using orm or self.orm.

SQLAState.new_session(self, *, orm=None): Context manager to create a new session from orm or self.orm.

• using_session(orm=None, session=None): A context manager to prepare an SQLAlchemy session for use by a suite.

  Parameters:

  • orm: optional reference ORM, an object with a .session() method for creating a new session. Default: if needed, obtained from the global state.orm.
  • session: optional existing session. Default: the global state.session if not None, otherwise created by orm.session().

  If a new session is created, the new session and reference ORM are pushed onto the globals state.session and state.orm respectively.

  If an existing session is reused, the suite runs within a savepoint from session.begin_nested().

• with_orm(function, *a, orm=None, **kw): Call function with the supplied orm in the shared state.

• with_session(function, *a, orm=None, session=None, **kw): Call function(*a,session=session,**kw), creating a session if required. The function function runs within a transaction, nested if the session already exists. If a new session is created it is set as the default session in the shared state.

  This is the inner mechanism of @auto_session and ORM.auto_session.

  Parameters:

  • function: the function to call
  • a: the positional parameters
  • orm: optional ORM class with a .session() context manager method such as the ORM base class supplied by this module.
  • session: optional existing ORM session
  • kw: other keyword arguments, passed to function

  One of orm or session must be not None; if session is None then one is made from orm.session() and used as a context manager.

  The session is also passed to function as the keyword parameter session to support nested calls.

Release Log

Release 20260531:

• Expose the db URL normalisation stuff as methods.
• ORM.startup_shutdown: do not obtain the db lock, we will now do this per orchestrated_session().
• New ORM.dbshell() method to run an interactive db prompt.

Release 20250308: ORM.startup_shutdown: make the poll interval 2s instead of 0.2s, which seems a bit aggressive when there's contention.

Release 20241122:

• RelationProxy: getitem now raises KeyError, getattr now raises AttributeError.
• RelationProxy: new optional "missing" parameter to provide a callable for field values when there is no matching record in the relation.

Release 20241005: ORM.init: small bugfix.

Release 20240723: ORM: run self.Base.metadata.create_all() on the first use of the db.

Release 20230612:

• Use cs.fileutils.lockfile context manager instead of makelockfile.
• Rename arranged_session to orchestrated_session.
• Some tweaks around connection closes, still edging towards a good work practice for easily doing short lived stuff (for cs.ebooks.calibre).

Release 20230212: ORM.init: drop case_sensitive, no longer supported?

Release 20220606:

• BasicTableMixin: provide DEFAULT_ID_COLUMN='id', by_id() has new optional id_column parameter.
• RelationProxy factory to make base classes which proxy a relation, for circumstances where you want to minimise access to the db itself.
• ORM.engine_keywords: turn on echo mode only if "SQL" in $DEBUG.

Release 20220311: Many updates and small fixes.

Release 20210420:

• ORM: drop .Session from docstring, no longer used.
• Rename ORM.sessionmaker to ORM._sessionmaker, not for public use.
• ORM: replace session with arranged_session, which allocates a session in conformance with ORM.serial_sessions (serial sessions are used with SQLite).
• Drop @ORM.auto_session and @ORM.orm_method decorators, no longer used.
• SQLAState.new_session: use orm.arranged_session(), use begin_nested(); SQLAState.auto_session: use begin_nested().

Release 20210322: Delete escaped debug code which issued a RuntimeError.

Release 20210321:

• Default session support, particularly though an ORM's .sqla_state per-Thread state object - this allows removal of a lot of plumbing and @auto_session decoration.
• Support for serialised sessions, for db backend where only one session may be active at a time; this brings easy support for multithreaded SQLite access.

Release 20210306:

• Rename _state to state, making it public.
• Some other internal changes.

Release 20201025:

• New BasicTableMixin and HasIdMixin classes with useful methods and a typical id Column respectively.
• Assorted fixes and improvements.

Release 20190830.1: Have the decorators set .module.

Release 20190830: @json_column: small docstring improvement.

Release 20190829:

• Bugfix @json_column setter: mark the column as modified for the ORM.
• New push_log_level context manager and @log_level decorator to temporarily change the SQLAlchemy logging handler level.

Release 20190812:

• Make ORM a MultiOpenMixin.
• get_json_field: use forgotten default parameter.
• Other minor changes.

Release 20190526:

• Support for virtual columns mapped to a JSON column interior value:
• New functions find_json_field, get_json_field, set_json_field.
• New decorator @json_column for declaritive_base tables.

Release 20190517:

• Make ORM._Session private session factory the public ORM.Session factory for external use.
• with_session: preexisting sessions still trigger a session.begin_nested, removes flush/commit tension elsewhere.

Release 20190403:

• Rename @ORM.orm_auto_session to @ORM.auto_session.
• New @orm_auto_session decorator for methods of objects with a .orm attribute.

Release 20190319.1: Initial release. ORM base class, @auto_session decorator.