zensols.db package

Submodules

zensols.db.bean

Basic CRUD utility classes

class zensols.db.bean.Bean

Bases: ABC

A container class like a Java bean.

__init__()

get_attr_names()

Return a list of string attribute names.

Return type:

Tuple[str]

get_attrs()

Return a dict of attributes that are meant to be persisted.

Return type:

Dict[str, Any]

get_insert_row()

Return a row of data meant to be inserted into the database. This method implementation leaves off the first attriubte assuming it contains a unique (i.e. row ID) of the object. See get_row().

Return type:

Tuple[Any]

get_row()

Return a row of data meant to be printed. This includes the unique ID of the bean (see get_insert_row()).

Return type:

Tuple[Any]

class zensols.db.bean.BeanDbPersister(conn_manager, sql_file=None, row_factory='tuple', select_name=None, select_by_id_name=None, select_exists_name=None, insert_name=None, update_name=None, delete_name=None, keys_name=None, count_name=None)

Bases: UpdatableBeanDbPersister

A class that contains the remaining CRUD funtionality the super class doesn’t have.

__init__(conn_manager, sql_file=None, row_factory='tuple', select_name=None, select_by_id_name=None, select_exists_name=None, insert_name=None, update_name=None, delete_name=None, keys_name=None, count_name=None)

count_name: str = None

The name of the SQL entry used to get a row count.

get_count()

Return the number of rows in the bean table.

Return type:

int

get_keys()

Return the unique keys from the bean table.

Return type:

Iterable[Any]

keys_name: str = None

The name of the SQL entry used to fetch all keys.

class zensols.db.bean.ConnectionManager

Bases: ABC

Instance DB-API connection lifecycle.

__init__()

abstract create()

Create a connection to the database.

Return type:

Any

dispose(conn)

Close the connection to the database.

abstract drop()

Remove all objects from the database or the database itself.

For SQLite, this deletes the file. In database implementations, this might drop all objects from the database. Regardless, it is expected that create is able to recreate the database after this action.

Return type:

bool

Returns:

whether the database was dropped

execute(conn, sql, params, row_factory, map_fn)

Execute SQL on a database connection.

The row_factory tells the method how to interpret the row data in to an object that’s returned. It can be one of:

• tuple: tuples (the default)

• identity: return the unmodified form from the database

• dict: for dictionaries

• pandas: for a pandas.DataFrame

• otherwise: a function or class

Compare this with map_fn, which transforms the data that’s given to the row_factory.

Parameters:

• conn (Any) – the connection object with the database

• sql (str) – the string SQL to execute

• params (Tuple[Any, ...]) – the parameters given to the SQL statement (populated with ?) in the statement

• row_factory (Union[str, Callable]) – tuple, dict, pandas or a function

• map_fn (Callable) – a function that transforms row data given to the row_factory

See:

DbPersister.execute().

Return type:

Tuple[Union[dict, tuple, DataFrame]]

execute_no_read(conn, sql, params)

Return database level information such as row IDs rather than the results of a query. Use this when inserting data to get a row ID.

Parameters:

• conn (Any) – the connection object with the database

• sql (str) – the SQL statement used on the connection’s cursor

• params (Tuple[Any, ...]) – the parameters given to the SQL statement (populated with ?) in the statement

See:

DbPersister.execute_no_read().

Return type:

int

insert_rows(conn, sql, rows, errors, set_id_fn, map_fn)

Insert a tuple of rows in the database and return the current row ID.

Parameters:

• rows (Iterable[Any]) – a sequence of tuples of data (or an object to be transformed, see map_fn in column order of the SQL provided by the entry insert_name

• errors (str) – if this is the string raise then raise an error on any exception when invoking the database execute, otherwise use ignore to ignore errors

• set_id_fn (Callable) – a callable that is given the data to be inserted and the row ID returned from the row insert as parameters

• map_fn (Callable) – if not None, used to transform the given row in to a tuple that is used for the insertion

Return type:

int

Returns:

the rowid of the last row inserted

See InsertableBeanDbPersister.insert_rows().

register_persister(persister)

Register the persister used for this connection manager.

Parameters:

persister (DbPersister) – the persister used for connection management

exception zensols.db.bean.DBError

Bases: Exception

“Raised for all zensols.db` related errors.

__module__ = 'zensols.db.bean'

__weakref__

list of weak references to the object (if defined)

class zensols.db.bean.DbPersister(conn_manager, sql_file=None, row_factory='tuple')

Bases: object

CRUDs data to/from a DB-API connection.

__init__(conn_manager, sql_file=None, row_factory='tuple')

conn_manager: ConnectionManager

Used to create DB-API connections.

execute(**kwargs) → Tuple[dict | tuple | pd.DataFrame]

Execute SQL on a database connection.

The row_factory tells the method how to interpret the row data in to an object that’s returned. It can be one of:

• tuple: tuples (the default)

• dict: for dictionaries

• pandas: for a pandas.DataFrame

• otherwise: a function or class

Compare this with map_fn, which transforms the data that’s given to the row_factory.

Parameters:

• sql – the string SQL to execute

• params – the parameters given to the SQL statement (populated with ?) in the statement

• row_factory – tuple, dict, pandas or a function

• map_fn – a function that transforms row data given to the row_factory

Return type:

Tuple[Union[dict, tuple, pd.DataFrame]]

execute_by_name(name, params=(), row_factory=None, map_fn=None)

Just like execute() but look up the SQL statement to execute on the database connection.

The row_factory tells the method how to interpret the row data in to an object that’s returned. It can be one of:

• tuple: tuples (the default)

• dict: for dictionaries

• pandas: for a pandas.DataFrame

• otherwise: a function or class

Compare this with map_fn, which transforms the data that’s given to the row_factory.

Parameters:

• name (str) – the named SQL query in the sql_file

• params (Tuple[Any]) – the parameters given to the SQL statement (populated with ?) in the statement

• row_factory (Union[str, Callable]) – tuple, dict, pandas or a function

• map_fn (Callable) – a function that transforms row data given to the row_factory

See:

execute()

execute_no_read(**kwargs) → int

Just like execute_by_name(), but return database level information such as row IDs rather than the results of a query. Use this when inserting data to get a row ID.

Parameters:

• entry_name – the key in the SQL file whose value is used as the statement

• capture_rowid – if True, return the last row ID from the cursor

See:

execute_sql_no_read()

Return type:

int

execute_singleton_by_name(*args, **kwargs)

Just like execute_by_name() except return only the first item or None if no results.

execute_sql_no_read(**kwargs) → int

Execute SQL and return the database level information such as row IDs rather than the results of a query. Use this when inserting data to get a row ID.

Return type:

int

property metadata: Dict[str, str]

Return the metadata associated with the SQL file.

row_factory: Union[str, Type] = 'tuple'

The default method by which data is returned from execute_* methods.

See:

execute().

property sql_entries: Dict[str, str]

Return a dictionary of names -> SQL statements from the SQL file.

sql_file: Path = None

The text file containing the SQL statements (see DynamicDataParser).

class zensols.db.bean.InsertableBeanDbPersister(conn_manager, sql_file=None, row_factory='tuple', select_name=None, select_by_id_name=None, select_exists_name=None, insert_name=None)

Bases: ReadOnlyBeanDbPersister

A class that contains insert funtionality.

__init__(conn_manager, sql_file=None, row_factory='tuple', select_name=None, select_by_id_name=None, select_exists_name=None, insert_name=None)

insert(bean)

Insert a bean using the order of the values given in Bean.get_insert_row() as that of the SQL defined with entry insert_name given in the initializer.

Return type:

int

insert_beans(beans, errors='raise')

Insert a bean using the order of the values given in Bean.get_insert_row() as that of the SQL defined with entry insert_name given in the initializer.

Return type:

int

insert_dataframe(**kwargs) → int

Like insert_rows() but the data is taken a Pandas dataframe.

Parameters:

• df – the dataframe from which the rows are drawn

• set_id_fn – a callable that is given the data to be inserted and the row ID returned from the row insert as parameters

• map_fn – if not None, used to transform the given row in to a tuple that is used for the insertion

• chuck_size – the number of rows inserted at a time, so the number of interactions with the database are at most the row count of the dataframe / ``chunk_size``x

Return type:

int

Returns:

the rowid of the last row inserted

insert_name: str = None

The name of the SQL entry used to insert data/class instance.

insert_row(*row)

Insert a row in the database and return the current row ID.

Parameters:

row – a sequence of data in column order of the SQL provided by the entry insert_name

Return type:

int

insert_rows(**kwargs) → int

Insert a tuple of rows in the database and return the current row ID.

Parameters:

• rows – a sequence of tuples of data (or an object to be transformed, see map_fn in column order of the SQL provided by the entry insert_name

• errors – if this is the string raise then raise an error on any exception when invoking the database execute

• set_id_fn – a callable that is given the data to be inserted and the row ID returned from the row insert as parameters

• map_fn – if not None, used to transform the given row in to a tuple that is used for the insertion

Return type:

int

Returns:

the rowid of the last row inserted

class zensols.db.bean.ReadOnlyBeanDbPersister(conn_manager, sql_file=None, row_factory='tuple', select_name=None, select_by_id_name=None, select_exists_name=None)

Bases: DbPersister

A read-only persister that CRUDs data based on predefined SQL given in the configuration. The class optionally works with instances of Bean when row_factory is set to the target bean class.

__init__(conn_manager, sql_file=None, row_factory='tuple', select_name=None, select_by_id_name=None, select_exists_name=None)

exists(id)

Return True if there is a object with unique ID (or row ID) in the database. Otherwise return False.

Return type:

bool

get()

Return using the SQL provided by the entry identified by select_name.

Return type:

list

get_by_id(id)

Return an object using it’s unique ID, which is could be the row ID in SQLite.

select_by_id_name: str = None

The name of the SQL entry used to select a single row by unique ID.

select_exists_name: str = None

The name of the SQL entry used to determine if a row exists by unique ID.

select_name: str = None

The name of the SQL entry used to select data/class.

class zensols.db.bean.UpdatableBeanDbPersister(conn_manager, sql_file=None, row_factory='tuple', select_name=None, select_by_id_name=None, select_exists_name=None, insert_name=None, update_name=None, delete_name=None)

Bases: InsertableBeanDbPersister

A class that contains the remaining CRUD funtionality the super class doesn’t have.

__init__(conn_manager, sql_file=None, row_factory='tuple', select_name=None, select_by_id_name=None, select_exists_name=None, insert_name=None, update_name=None, delete_name=None)

delete(id)

Delete a row by ID.

Return type:

int

delete_name: str = None

The name of the SQL entry used to delete data/class instance(s).

update(bean)

Update a a bean that using the id attribute and its attributes as values.

Return type:

int

update_name: str = None

The name of the SQL entry used to update data/class instance(s).

update_row(*row)

Update a row using the values of the row with the current unique ID as the first element in *rows.

Return type:

int

class zensols.db.bean.connection

Bases: resource

Annotation used to create and dispose of DB-API connections.

__init__()

Create the instance based annotation.

Parameters:

• create_method_name – the name of the method that allocates

• destroy_method_name – the name of the method that deallocates

class zensols.db.bean.cursor(persister, sql=None, name=None, params=())

Bases: object

Iterate through rows of a database. The connection is automatically closed once out of scope.

Example:

config_factory: ConfigFactory = ...
persister: DbPersister = config_factory.instance('person_db_persister')
with cursor(persister, name='select_people') as c:
    for row in c:
        print(row)

__init__(persister, sql=None, name=None, params=())

Initialize with either name or sql (only one should be None).

Parameters:

• persister (DbPersister) – used to execute the SQL and obtain the cursor

• sql (str) – the string SQL to execute

• name (str) – the named SQL query in the DbPersister.sql_file

• params (Tuple[Any, ...]) – the parameters given to the SQL statement (populated with ?) in the statement

zensols.db.dataclass

Contains utility classes to persist dataclasses.dataclass.

class zensols.db.dataclass.DataClassDbPersister(conn_manager, sql_file=None, row_factory='tuple', select_name=None, select_by_id_name=None, select_exists_name=None, insert_name=None, update_name=None, delete_name=None, keys_name=None, count_name=None, bean_class=None)

Bases: BeanDbPersister

Persists instances of dataclasses.dataclass by narrowing the columns from select statements. Instead of select *, use select ${cols} in the SQL resource file.

See:

DataClassDynamicDataParser

__init__(conn_manager, sql_file=None, row_factory='tuple', select_name=None, select_by_id_name=None, select_exists_name=None, insert_name=None, update_name=None, delete_name=None, keys_name=None, count_name=None, bean_class=None)

bean_class: Type[dataclass] = None

The data class that is CRUD’d for DB operations.

class zensols.db.dataclass.DataClassDynamicDataParser(dd_path, bean_class)

Bases: DynamicDataParser

An SQL data parser that replaces ${cols} in the SQL file with the dataclasses.dataclass fields.

See:

DataClassDbPersister

ID_FIELD: ClassVar[str] = 'id'

The name of the column that has the unique identifier of the row/object.

__init__(dd_path, bean_class)

Initialize.

Parameters:

dd_path (Path) – the path of the file to parse

zensols.db.parse

Simple parsing of a dynamic definition langauge (DDL) and dynamic manipulation language (DML) files.

class zensols.db.parse.DynamicDataParser(dd_path)

Bases: object

Parse a DDL/DML file meant also for prototyping.

For example the file:

-- meta=init_sections=create_tables,create_idx
-- name=create_idx
create index person_name on person(name);

-- name=create_tables
create table person (id int, name text, age int);

Would have create_idx and create_tables as sections and meta data:

{'init_sections':
 'create_tables,create_idx'}

COMMENT_PAT: ClassVar[Pattern] = re.compile('^--.*')

META_PAT: ClassVar[Pattern] = re.compile('^-- meta=([a-zA-Z0-9_]+)=(.+)$')

SEC_START_PAT: ClassVar[Pattern] = re.compile('^-- name=([a-zA-Z0-9_]+)')

__init__(dd_path)

Initialize.

Parameters:

dd_path (Path) – the path of the file to parse

get_init_db_sqls()

Return the set of statements that create all DB objects needed to fully CRUD.

Return type:

Iterable[str]

property metadata: Dict[str, str]

Return the meta data found int he parse object.

property sections: Dict[str, str]

Return the sections of the file.

zensols.db.sqlite

Convenience wrapper for the Python DB-API library, and some specificly for the SQLite library.

class zensols.db.sqlite.SqliteAttachConnectionManager(db_file, database_name)

Bases: ConnectionManager

An SQLite connection factory that attaches a file as a database.

__init__(db_file, database_name)

create()

Create a connection as an attached database to a file.

Return type:

Connection

database_name: str

The name of the database used to attach to db_file.

db_file: Path

The SQLite database file to read or create.

drop()

Dropping a memory SQLite connection is not supported.

Return type:

bool

class zensols.db.sqlite.SqliteConnectionManager(db_file, create_db=True)

Bases: ConnectionManager

An SQLite connection factory.

__init__(db_file, create_db=True)

create()

Create a connection by accessing the SQLite file.

Raises:

DBError – if the SQLite file does not exist (caveat see :obj:create_db)

Return type:

Connection

create_db: bool = True

If True, create the database if it does not already exist. Otherwise, DBError is raised (see create()).

db_file: Path

The SQLite database file to read or create.

drop()

Delete the SQLite database file from the file system.

Return type:

bool

class zensols.db.sqlite.SqliteDbStash(encoder_decoder=<factory>, path=None)

Bases: DbStash

A Stash implementation that uses an SQLite database to store data.

__init__(encoder_decoder=<factory>, path=None)

path: Path = None

The directory of where to store the files.

zensols.db.stash

Adapt a database centric DbPersister to a general Stash container.

class zensols.db.stash.AlternateKeyBeanStash(persister, key_to_id_name, keys_name)

Bases: BeanStash

A stash that uses another key rather than some unique primary key (i.e. rowid for SQLite). It does this by looking up the alternate key in some other column and resolves to the unique primary key.

The domain and range of the function (_key_to_id()) that maps alternate keys to unique primary keys ate strings.

_key_to_id(name)

Maps alternate keys to unique primary keys.

Parameters:

name (str) – the alternate key, which is usually a more client friendly string

Return type:

Optional[str]

Returns:

the unique primary key in the database (usually an int)

__init__(persister, key_to_id_name, keys_name)

delete(name)

Delete the resource for data pointed to by name or the entire resource if name is not given.

dump(name, inst)

Since this implementation can let the database auto-increment the unique/primary key, beware of “changing” keys.

Raises:

DBError – if the key changes after inserted it will raise a DBError; for this reason, it’s best to pass None as name

exists(name)

Return True if data with key name exists.

Implementation note: This Stash.exists() method is very inefficient and should be overriden.

Return type:

bool

key_to_id_name: str

The select method SQL name that selects the unique priamry to the alterante key.

keys()

Return an iterable of keys in the collection.

Return type:

Iterable[str]

keys_name: str

The select method SQL name that selects the alternate in keys().

load(name)

Load a data value from the pickled data with key name. Semantically, this method loads the using the stash’s implementation. For example DirectoryStash loads the data from a file if it exists, but factory type stashes will always re-generate the data.

See:

get()

Return type:

Any

class zensols.db.stash.BeanStash(persister)

Bases: Stash

A stash that uses a backing DB-API backed BeanDbPersister.

__init__(persister)

delete(name)

Delete the resource for data pointed to by name or the entire resource if name is not given.

dump(name, inst)

Since this implementation can let the database auto-increment the unique/primary key, beware of “changing” keys.

Raises:

DBError – if the key changes after inserted it will raise a DBError; for this reason, it’s best to pass None as name

exists(name)

Return True if data with key name exists.

Implementation note: This Stash.exists() method is very inefficient and should be overriden.

Return type:

bool

keys()

Return an iterable of keys in the collection.

Return type:

Iterable[str]

load(name)

Load a data value from the pickled data with key name. Semantically, this method loads the using the stash’s implementation. For example DirectoryStash loads the data from a file if it exists, but factory type stashes will always re-generate the data.

See:

get()

Return type:

Any

persister: BeanDbPersister

The delegate bean persister.

class zensols.db.stash.DbStash(encoder_decoder=<factory>)

Bases: Stash

A relational database to store stash keys and values. It creates a single table with only two columns: one for the (string) key and the other for the values.

__init__(encoder_decoder=<factory>)

clear()

Delete all data from the from the stash.

Important: Exercise caution with this method, of course.

delete(name)

Delete the resource for data pointed to by name or the entire resource if name is not given.

dump(name, inst)

Since this implementation can let the database auto-increment the unique/primary key, beware of “changing” keys.

Raises:

DBError – if the key changes after inserted it will raise a DBError; for this reason, it’s best to pass None as name

encoder_decoder: DbStashEncoderDecoder

Used to encode and decode the data with the SQLite database. To use binary data, set this to an instance of

This should be set to:

• DbStashEncoderDecoder: store text values

• PickleDbStashEncoderDecoder: store binary data (default)

• jsonpickle: store JSON (needs pip install jsonpickle); use encoder_decoder = eval({'import': ['jsonpickle']}): jsonpickle in application configurations

You can write your own by extending DbStashEncoderDecoder.

exists(name)

Return True if data with key name exists.

Implementation note: This Stash.exists() method is very inefficient and should be overriden.

Return type:

bool

keys()

Return an iterable of keys in the collection.

Return type:

Iterable[str]

load(name)

Load a data value from the pickled data with key name. Semantically, this method loads the using the stash’s implementation. For example DirectoryStash loads the data from a file if it exists, but factory type stashes will always re-generate the data.

See:

get()

Return type:

Any

property persister: BeanDbPersister

The persister used to interface with the database.

class zensols.db.stash.DbStashEncoderDecoder

Bases: object

Encodes and decodes data for SqliteStash.

decode(data)

Return type:

Any

encode(data)

Return type:

Union[str, bytes]

class zensols.db.stash.PickleDbStashEncoderDecoder

Bases: DbStashEncoderDecoder

An implementation that encodes and decodes using pickle.

decode(data)

Return type:

Any

encode(data)

Return type:

Union[str, bytes]

Module contents