table

Description

This module provides a set of classes for working with database tables in an object-oriented manner.

The module includes classes for representing table records, columns, and different types of tables (memory-based and SQL-based). It also provides utility classes for generating column values and handling table selection predicates and offers an abstraction layer that allows for consistent interaction with different types of tables.

This module is designed to work with asynchronous operations and provides a flexible and extensible framework for database operations.

Warning

Never create table instances outside of an asynchronous context (except when you’ve re-implemented their logic). This is because when creating a table, it needs an active event loop. You can use lazy initialization instead:

from sl3aio.table import MemoryTable

class Database:
    my_table: MemoryTable

    @classmethod
    def setup(cls) -> None:
        cls.my_table = MemoryTable("my_table", ...)

async def main() -> None:
    Database.setup()

Key Components

Other Components

  • TableSelectionPredicate: Protocol for defining predicates used in table selection operations.

  • Table: Abstract base class for all table types.

  • SqlTable: Abstract base class for SQL-based tables.

Usage Examples

from sl3aio.table import MemoryTable, TableColumn

# Define columns
id_column = TableColumn("id", "INTEGER", primary=True)
name_column = TableColumn("name", "TEXT", nullable=False)
age_column = TableColumn("age", "INTEGER", default=0)

# Create a MemoryTable
person_table = MemoryTable("persons", (id_column, name_column, age_column))

# Insert a record
await person_table.insert(id=1, name="Alice", age=30)

# Select records
async for record in person_table.select(lambda r: r.age > 25):
    print(record.name, record.age)

# Update a record
await person_table.update(lambda r: r.id == 1, age=31)

# Delete a record
await person_table.delete(lambda r: r.name == "Alice")

from sl3aio.table import SolidTable, TableColumn
from sl3aio.executor import ConnectionManager

# Define columns
id_column = TableColumn("id", "INTEGER", primary=True)
title_column = TableColumn("title", "TEXT", nullable=False)
author_column = TableColumn("author", "TEXT", nullable=False)

# Create a ConnectionManager
conn_manager = ConnectionManager("path/to/database.db")

# Create a SolidTable
book_table = SolidTable("books", (id_column, title_column, author_column), conn_manager)

# Create the table in the database
await book_table.create()

# Insert a record
await book_table.insert(id=1, title="1984", author="George Orwell")

# Select records
async for record in book_table.select(lambda r: r.author == "George Orwell"):
    print(record.title)

# Update a record
await book_table.update(lambda r: r.id == 1, title="Nineteen Eighty-Four")

# Delete a record
await book_table.delete(lambda r: r.title == "Nineteen Eighty-Four")

# Drop the table
await book_table.drop()

from sl3aio.table import TableColumnValueGenerator, TableColumn, SolidTable
from random import randint

# Define a generator for random IDs
@TableColumnValueGenerator.from_function("random_id")
def random_id():
    return randint(1000, 9999)

# Create a column with the generator
id_column = TableColumn("id", "INTEGER", generator=TableColumnValueGenerator.get_by_name("random_id"))
name_column = TableColumn("name", "TEXT", nullable=False)

# Create a table with the generated column
user_table = SolidTable("users", (id_column, name_column), conn_manager)

# Insert a record (id will be generated automatically)
await user_table.insert(name="Bob")

from sl3aio.table import TableSelectionPredicate

# Define a custom predicate
class AgeRangePredicate(TableSelectionPredicate):
    def __init__(self, min_age: int, max_age: int):
        self.min_age = min_age
        self.max_age = max_age

    async def __call__(self, record):
        return self.min_age <= record.age <= self.max_age

# Use the custom predicate
age_range = AgeRangePredicate(25, 35)
async for record in person_table.select(age_range):
    print(record.name, record.age)

See also

easytable

Convinient and easy interface to work with tables.

executor

Core module of this library.

class sl3aio.table.TableRecord(*args: T, **kwargs: T)[source]

Bases: tuple[T, …], Generic

Represents a single record in a table.

This class extends the built-in tuple class to provide additional functionality specific to table records.

Note

Hash of the record is same as the hash of the tuple containing only values of unique/primary (nonrepeating) columns. The equality operator works similarly. So, as in the sqlite, two records of the table, which doesn’t have unique/primary columns, are always different.

table: ClassVar[Table]

The table to which this record belongs.

nonrepeating: ClassVar[tuple[str, ...]]

Names of columns that are unique or primary keys.

fields: ClassVar[tuple[str, ...]]

Names of all columns in the table.

executor: ClassVar[Executor]

An executor for running operations on the record.

static __new__(cls, *args: T, **kwargs: T) → TableRecord[source]

Create a new instance of the record. Same as TableRecord.make() but synchronous.

classmethod make_subclass(table: Table, *columns: TableColumn) → type[TableRecord][source]

Create a new subclass of TableRecord for a specific table.

Parameters:

  • table (Table [T]) – The table for which to create the subclass.

  • *columns (TableColumn [T]) – The columns of the table.

Returns:

A new subclass of TableRecord.

Return type:

type [TableRecord [T]]

async classmethod make(*args: T, **kwargs: T) → TableRecord[source]

Asynchronously create a new instance of the record.

The positional arguments must be in the order of the columns in the table. The names of the keyword arguments must match the names of the columns in the table. If a value is not passed for some columns, value, obtained from TableColumn.get_default(), will be used.

Note

You must provide values for every column in the table that can’t be null and doesn’t have a default value.

Parameters:

  • *args (T) – Positional arguments for the record values.

  • **kwargs (T) – Keyword arguments for the record values.

Returns:

A new instance of the record.

Return type:

TableRecord [T]

asdict() → dict[str, T][source]

Convert the record to a dictionary.

Returns:

A dictionary representation of the record.

Return type:

dict [str, T]

astuple() → tuple[T, ...][source]

Convert the record to a tuple.

Returns:

A tuple representation of the record.

Return type:

tuple [T, …]

async replace(**to_replace: T) → Self[source]

Create a new record with some values replaced.

Parameters:

**to_replace (T) – The values to replace in the new record.

Returns:

A new record with the specified values replaced.

Return type:

Self

count(value, /)

Return number of occurrences of value.

class sl3aio.table.TableSelectionPredicate[source]

Bases: Protocol, Generic

Protocol for defining predicates used in table selection operations.

This protocol defines the interface for callable objects that can be used to filter records in table operations.

async __call__(record: TableRecord) → bool[source]

Evaluate the predicate for a given record.

Parameters:

record (TableRecord [T]) – The record to evaluate on.

Returns:

True if the record matches the predicate, False otherwise.

Return type:

bool

class sl3aio.table.TableColumnValueGenerator(name: str, generator: Callable[[], T | Awaitable])[source]

Bases: Generic

Generates values for table columns.

This class is used to create generators for column values, which can be used as default values or for generating values during insert operations.

See also

TableColumn

name: str

The name of the generator.

generator: Callable[[], T | Awaitable]

The function that generates values for the column. Can be both async and sync.

classmethod make(name: str, generator: Callable[[], T | Awaitable], register: bool = True) → TableColumnValueGenerator[source]

Create a new TableColumnValueGenerator instance.

Parameters:

  • name (str) – The name of the generator.

  • generator (Callable [[], T | Awaitable [T]]) – The function that generates values.

  • register (bool, optional) – Whether to register the generator globally. Defaults to True.

Returns:

A new instance of TableColumnValueGenerator.

Return type:

TableColumnValueGenerator [T]

classmethod from_function(name: str, register: bool = True) → Callable[[Callable[[], T | Awaitable]], TableColumnValueGenerator][source]

Decorator to create a TableColumnValueGenerator from a function.

Parameters:

  • name (str) – The name of the generator.

  • register (bool, optional) – Whether to register the generator globally. Defaults to True.

Returns:

A decorator that creates a TableColumnValueGenerator.

Return type:

callable

classmethod get_by_name(name: str) → TableColumnValueGenerator | None[source]

Retrieve a registered generator by name.

Parameters:

name (str) – The name of the generator to retrieve.

Returns:

The registered generator, or None if not found.

Return type:

TableColumnValueGenerator | None

copy() → Self[source]

Create a copy of the generator.

Returns:

A new instance of the generator with the same attributes.

Return type:

Self

register() → Self[source]

Register the copy of generator for global use.

Returns:

Self for chaining.

Return type:

Self

unregister() → Self[source]

Unregister the generator.

Returns:

Self for chaining.

Return type:

Self

__next__() → T[source]

Retrieve a next value for the column.

Returns:

Generated value.

Return type:

T

class sl3aio.table.TableColumn(name: str, typename: str, default: T | None = None, generator: TableColumnValueGenerator | None = None, primary: bool = False, unique: bool = False, nullable: bool = True)[source]

Bases: Generic

Represents a column in a table.

This class defines the properties of a table column, including its name, data type, and constraints.

Note

If both the generator and the default parameters are specified, preference is given to generator.

See also

TableColumnValueGenerator, Table

name: str

The name of the column.

typename: str

The data type of the column.

default: T | None

The default value for the column. Defaults to None.

generator: TableColumnValueGenerator | None

A generator for column values. Defaults to None.

primary: bool

Whether this column is a primary key. Defaults to False.

unique: bool

Whether this column has a unique constraint. Defaults to False.

nullable: bool

Whether this column can contain NULL values. Defaults to True.

classmethod from_sql(sql: str, default: T | TableColumnValueGenerator | None = None) → TableColumn[source]

Create a TableColumn instance from an SQL column definition.

Note

Columns, that generated by sqlite using GENERATED keyword, won’t be interpreted correctly. To make column generated, you need to create generator for it using the TableColumnValueGenerator and then set column DEFAULT value to "$Generated:generator_name"

Parameters:

  • sql (str) – The SQL definition of the column.

  • default (T | None, optional) – The default value or default value generator for the column.

Returns:

A new TableColumn instance.

Return type:

TableColumn [T]

to_sql() → str[source]

Generate the SQL representation of the column.

Note

  • If column’s typename is not present in Parser registry, it would be represented as TEXT.

  • Columns with specified generator would have their DEFAULT value setted to "$Generated:generator_name".

Returns:

The SQL definition of the column.

Return type:

str

get_default() → T | None[source]

Get the default value for the column.

First this method will try to get the default value from generator. If it not specified it will use the default value.

Returns:

The default value for the column.

Return type:

T | None

class sl3aio.table.Table(name: str, _columns: tuple[TableColumn, ...])[source]

Bases: ABC, Generic

Abstract base class for all table types.

This class defines the common interface and functionality for different types of tables (e.g., in-memory tables, SQL tables).

See also

TableColumn, TableRecord, MemoryTable, SolidTable

name: str

The name of the table.

property columns: tuple[TableColumn, ...]

The columns of the table.

async make_record(*args: T, **kwargs: T) → TableRecord[source]

Create a new record for this table.

For more details, see TableRecord.make().

Parameters:

  • *args (T) – Positional arguments for the record values.

  • **kwargs (T) – Keyword arguments for the record values.

Returns:

A new record for this table.

Return type:

TableRecord [T]

async start_executor() → None[source]

Start the table’s executor.

async stop_executor() → None[source]

Stop the table’s executor.

abstract async length() → int[source]

Get the number of records in the table.

Returns:

The number of records in the table.

Return type:

int

abstract async contains(record: TableRecord) → bool[source]

Check if a record exists in the table.

Parameters:

record (TableRecord [T]) – The record to check for.

Returns:

True if the record exists in the table, False otherwise.

Return type:

bool

abstract async insert(ignore_existing: bool = False, **values: T) → TableRecord[source]

Insert a new record into the table.

Parameters:

  • ignore_existing (bool, optional) – Whether to ignore existing records. Defaults to False.

  • **values (T) – The values for the new record.

Returns:

The inserted record.

Return type:

TableRecord [T]

See also

Table.insert_many()

async insert_many(ignore_existing: bool = False, *values: dict[str, T]) → AsyncIterator[TableRecord][source]

Insert multiple records into the table.

Important

You must iterate over the result for operation to be performed.

Parameters:

  • ignore_existing (bool, optional) – Whether to ignore existing records. Defaults to False.

  • *values (dict [str, T]) – Dictionaries containing the values for each record to insert.

Yields:

TableRecord [T] – The inserted records.

See also

Table.insert()

abstract select(predicate: TableSelectionPredicate | None = None) → AsyncIterator[TableRecord][source]

Select records from the table.

Note

If predicate isn’t specified, yields the whole table.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records. Defaults to None.

Yields:

TableRecord [T] – The selected records.

See also

Table.select_one()

async select_one(predicate: TableSelectionPredicate | None = None) → TableRecord | None[source]

Select a single record from the table.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records. Defaults to None.

Returns:

The selected record, or None if no record matches.

Return type:

TableRecord [T] | None

See also

Table.select()

async count(predicate: TableSelectionPredicate | None = None) → int[source]

Count the number of records in the table that match the predicate.

Note

If no predicate is specified, the result will be the same as for the Table.length() method.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records. Defaults to None.

Returns:

Number of records that match the predicate.

Return type:

int

abstract deleted(predicate: TableSelectionPredicate | None = None) → AsyncIterator[TableRecord][source]

Remove and yield deleted records from the table.

Note

If predicate isn’t specified, yields the whole table and then clears it.

Important

You must iterate over the result for operation to be performed.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records to remove. Defaults to None.

Yields:

TableRecord [T] – The removed records.

See also

Table.delete(), Table.delete_one()

async delete(predicate: TableSelectionPredicate | None = None) → None[source]

Delete records from the table.

Note

If predicate isn’t specified, clears the table.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records to delete. Defaults to None.

See also

Table.deleted(), Table.delete_one()

async delete_one(predicate: TableSelectionPredicate | None = None) → TableRecord | None[source]

Delete a single record from the table.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the record to delete. Defaults to None.

Returns:

The deleted record, or None if no record matches.

Return type:

TableRecord [T] | None

See also

Table.delete(), Table.delete_one()

abstract updated(predicate: TableSelectionPredicate | None = None, **to_update: T) → AsyncIterator[TableRecord][source]

Update records in the table and yield the updated records.

Note

If predicate isn’t specified, updates and yields every record.

Important

You must iterate over the result for operation to be performed.

Parameters:

  • predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records to update. Defaults to None.

  • **to_update (T) – The values to update in the matching records.

Yields:

TableRecord [T] – The updated records.

See also

Table.update(), Table.update_one()

async update(predicate: TableSelectionPredicate | None = None, **to_update: T) → None[source]

Update records in the table without yielding the updated records.

Note

If predicate isn’t specified, upadates every record.

Parameters:

  • predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records to update. Defaults to None.

  • **to_update (T) – The values to update in the matching records.

See also

Table.updated(), Table.update_one()

async update_one(predicate: TableSelectionPredicate | None = None, **to_update: T) → TableRecord | None[source]

Update a single record in the table.

Parameters:

  • predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the record to update. Defaults to None.

  • **to_update (T) – The values to update in the matching record.

Returns:

The updated record, or None if no record matches.

Return type:

TableRecord [T] | None

See also

Table.updated(), Table.update()

async __aenter__() → Self[source]

Asynchronous context manager entry point.

Entries the executor’s context manager, allowing to interact with the table.

Returns:

The Table instance.

Return type:

Self

See also

ConsistentExecutor.__aenter__()

async __aexit__(*args) → None[source]

Asynchronous context manager exit point.

Exits the executor’s context manager and disables the table.

Parameters:

*args – Arguments passed to the exit method.

See also

ConsistentExecutor.__aexit__()

class sl3aio.table.MemoryTable(name: str, _columns: tuple[~sl3aio.table.TableColumn, ...], _records: set[~sl3aio.table.TableRecord] = <factory>)[source]

Bases: Table, Generic

A concrete implementation of Table for interacting with in-memory databases.

This class provides methods for performing CRUD (Create, Read, Update, Delete) operations on ‘memory tables’ (actually, just a python sets). It implements the abstract methods defined in Table class.

Warning

You must call MemoryTable.start_executor() or enter table’s async context using async with table construction before acessing the table, otherwise the program will await for the request to complete forever.

See also

TableColumn, TableRecord, Table

async length() → int[source]

Get the number of records in the table.

Returns:

The number of records in the table.

Return type:

int

async contains(record: TableRecord) → bool[source]

Check if a record exists in the table.

Parameters:

record (TableRecord [T]) – The record to check for.

Returns:

True if the record exists in the table, False otherwise.

Return type:

bool

async insert(ignore_existing: bool = False, **values: T) → TableRecord[source]

Insert a new record into the table.

Parameters:

  • ignore_existing (bool, optional) – Whether to ignore existing records. Defaults to False.

  • **values (T) – The values for the new record.

Returns:

The inserted record.

Return type:

TableRecord [T]

See also

Table.insert_many()

async select(predicate: TableSelectionPredicate | None = None) → AsyncIterator[TableRecord][source]

Select records from the table.

Note

If predicate isn’t specified, yields the whole table.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records. Defaults to None.

Yields:

TableRecord [T] – The selected records.

See also

Table.select_one()

async deleted(predicate: TableSelectionPredicate | None = None) → AsyncIterator[TableRecord][source]

Remove and yield deleted records from the table.

Note

If predicate isn’t specified, yields the whole table and then clears it.

Important

You must iterate over the result for operation to be performed.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records to remove. Defaults to None.

Yields:

TableRecord [T] – The removed records.

See also

Table.delete(), Table.delete_one()

async updated(predicate: TableSelectionPredicate | None = None, **to_update: T) → AsyncIterator[TableRecord][source]

Update records in the table and yield the updated records.

Note

If predicate isn’t specified, updates and yields every record.

Important

You must iterate over the result for operation to be performed.

Parameters:

  • predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records to update. Defaults to None.

  • **to_update (T) – The values to update in the matching records.

Yields:

TableRecord [T] – The updated records.

See also

Table.update(), Table.update_one()

async __aenter__() → Self

Asynchronous context manager entry point.

Entries the executor’s context manager, allowing to interact with the table.

Returns:

The Table instance.

Return type:

Self

See also

ConsistentExecutor.__aenter__()

async __aexit__(*args) → None

Asynchronous context manager exit point.

Exits the executor’s context manager and disables the table.

Parameters:

*args – Arguments passed to the exit method.

See also

ConsistentExecutor.__aexit__()

property columns: tuple[TableColumn, ...]

The columns of the table.

async count(predicate: TableSelectionPredicate | None = None) → int

Count the number of records in the table that match the predicate.

Note

If no predicate is specified, the result will be the same as for the Table.length() method.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records. Defaults to None.

Returns:

Number of records that match the predicate.

Return type:

int

async delete(predicate: TableSelectionPredicate | None = None) → None

Delete records from the table.

Note

If predicate isn’t specified, clears the table.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records to delete. Defaults to None.

See also

Table.deleted(), Table.delete_one()

async delete_one(predicate: TableSelectionPredicate | None = None) → TableRecord | None

Delete a single record from the table.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the record to delete. Defaults to None.

Returns:

The deleted record, or None if no record matches.

Return type:

TableRecord [T] | None

See also

Table.delete(), Table.delete_one()

async insert_many(ignore_existing: bool = False, *values: dict[str, T]) → AsyncIterator[TableRecord]

Insert multiple records into the table.

Important

You must iterate over the result for operation to be performed.

Parameters:

  • ignore_existing (bool, optional) – Whether to ignore existing records. Defaults to False.

  • *values (dict [str, T]) – Dictionaries containing the values for each record to insert.

Yields:

TableRecord [T] – The inserted records.

See also

Table.insert()

async make_record(*args: T, **kwargs: T) → TableRecord

Create a new record for this table.

For more details, see TableRecord.make().

Parameters:

  • *args (T) – Positional arguments for the record values.

  • **kwargs (T) – Keyword arguments for the record values.

Returns:

A new record for this table.

Return type:

TableRecord [T]

name: str

The name of the table.

async select_one(predicate: TableSelectionPredicate | None = None) → TableRecord | None

Select a single record from the table.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records. Defaults to None.

Returns:

The selected record, or None if no record matches.

Return type:

TableRecord [T] | None

See also

Table.select()

async start_executor() → None

Start the table’s executor.

async stop_executor() → None

Stop the table’s executor.

async update(predicate: TableSelectionPredicate | None = None, **to_update: T) → None

Update records in the table without yielding the updated records.

Note

If predicate isn’t specified, upadates every record.

Parameters:

  • predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records to update. Defaults to None.

  • **to_update (T) – The values to update in the matching records.

See also

Table.updated(), Table.update_one()

async update_one(predicate: TableSelectionPredicate | None = None, **to_update: T) → TableRecord | None

Update a single record in the table.

Parameters:

  • predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the record to update. Defaults to None.

  • **to_update (T) – The values to update in the matching record.

Returns:

The updated record, or None if no record matches.

Return type:

TableRecord [T] | None

See also

Table.updated(), Table.update()

class sl3aio.table.SqlTable(name: str, _columns: tuple[TableColumn, ...], _executor: ConnectionManager)[source]

Bases: Table, ABC, Generic

Abstract base class for SQL-based tables.

This class extends the functionality of the Table class to work with SQL databases. It provides methods for interacting with SQL tables and manages the connection to the database.

See also

SolidTable, TableColumn, TableRecord, Table

async __aenter__() → Self

Asynchronous context manager entry point.

Entries the executor’s context manager, allowing to interact with the table.

Returns:

The Table instance.

Return type:

Self

See also

ConsistentExecutor.__aenter__()

async __aexit__(*args) → None

Asynchronous context manager exit point.

Exits the executor’s context manager and disables the table.

Parameters:

*args – Arguments passed to the exit method.

See also

ConsistentExecutor.__aexit__()

property columns: tuple[TableColumn, ...]

The columns of the table.

abstract async contains(record: TableRecord) → bool

Check if a record exists in the table.

Parameters:

record (TableRecord [T]) – The record to check for.

Returns:

True if the record exists in the table, False otherwise.

Return type:

bool

async count(predicate: TableSelectionPredicate | None = None) → int

Count the number of records in the table that match the predicate.

Note

If no predicate is specified, the result will be the same as for the Table.length() method.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records. Defaults to None.

Returns:

Number of records that match the predicate.

Return type:

int

async delete(predicate: TableSelectionPredicate | None = None) → None

Delete records from the table.

Note

If predicate isn’t specified, clears the table.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records to delete. Defaults to None.

See also

Table.deleted(), Table.delete_one()

async delete_one(predicate: TableSelectionPredicate | None = None) → TableRecord | None

Delete a single record from the table.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the record to delete. Defaults to None.

Returns:

The deleted record, or None if no record matches.

Return type:

TableRecord [T] | None

See also

Table.delete(), Table.delete_one()

abstract deleted(predicate: TableSelectionPredicate | None = None) → AsyncIterator[TableRecord]

Remove and yield deleted records from the table.

Note

If predicate isn’t specified, yields the whole table and then clears it.

Important

You must iterate over the result for operation to be performed.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records to remove. Defaults to None.

Yields:

TableRecord [T] – The removed records.

See also

Table.delete(), Table.delete_one()

abstract async insert(ignore_existing: bool = False, **values: T) → TableRecord

Insert a new record into the table.

Parameters:

  • ignore_existing (bool, optional) – Whether to ignore existing records. Defaults to False.

  • **values (T) – The values for the new record.

Returns:

The inserted record.

Return type:

TableRecord [T]

See also

Table.insert_many()

async insert_many(ignore_existing: bool = False, *values: dict[str, T]) → AsyncIterator[TableRecord]

Insert multiple records into the table.

Important

You must iterate over the result for operation to be performed.

Parameters:

  • ignore_existing (bool, optional) – Whether to ignore existing records. Defaults to False.

  • *values (dict [str, T]) – Dictionaries containing the values for each record to insert.

Yields:

TableRecord [T] – The inserted records.

See also

Table.insert()

abstract async length() → int

Get the number of records in the table.

Returns:

The number of records in the table.

Return type:

int

async make_record(*args: T, **kwargs: T) → TableRecord

Create a new record for this table.

For more details, see TableRecord.make().

Parameters:

  • *args (T) – Positional arguments for the record values.

  • **kwargs (T) – Keyword arguments for the record values.

Returns:

A new record for this table.

Return type:

TableRecord [T]

name: str

The name of the table.

abstract select(predicate: TableSelectionPredicate | None = None) → AsyncIterator[TableRecord]

Select records from the table.

Note

If predicate isn’t specified, yields the whole table.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records. Defaults to None.

Yields:

TableRecord [T] – The selected records.

See also

Table.select_one()

async select_one(predicate: TableSelectionPredicate | None = None) → TableRecord | None

Select a single record from the table.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records. Defaults to None.

Returns:

The selected record, or None if no record matches.

Return type:

TableRecord [T] | None

See also

Table.select()

async start_executor() → None

Start the table’s executor.

async stop_executor() → None

Stop the table’s executor.

async update(predicate: TableSelectionPredicate | None = None, **to_update: T) → None

Update records in the table without yielding the updated records.

Note

If predicate isn’t specified, upadates every record.

Parameters:

  • predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records to update. Defaults to None.

  • **to_update (T) – The values to update in the matching records.

See also

Table.updated(), Table.update_one()

async update_one(predicate: TableSelectionPredicate | None = None, **to_update: T) → TableRecord | None

Update a single record in the table.

Parameters:

  • predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the record to update. Defaults to None.

  • **to_update (T) – The values to update in the matching record.

Returns:

The updated record, or None if no record matches.

Return type:

TableRecord [T] | None

See also

Table.updated(), Table.update()

abstract updated(predicate: TableSelectionPredicate | None = None, **to_update: T) → AsyncIterator[TableRecord]

Update records in the table and yield the updated records.

Note

If predicate isn’t specified, updates and yields every record.

Important

You must iterate over the result for operation to be performed.

Parameters:

  • predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records to update. Defaults to None.

  • **to_update (T) – The values to update in the matching records.

Yields:

TableRecord [T] – The updated records.

See also

Table.update(), Table.update_one()

property database: str

Get the name of the database this table belongs to.

async classmethod from_database(name: str, executor: ConnectionManager) → SqlTable[source]

Create a SqlTable instance from an existing database table.

Warning

Before awaiting this function, make sure that the executor is running, otherwise the program will freeze.

Parameters:

  • name (str) – The name of the existing table in the database.

  • executor (ConnectionManager) – The connection manager for the database.

Returns:

A new SqlTable instance representing the existing database table.

Return type:

SqlTable [T]

Raises:

AssertionError – If the table does not exist in the specified database.

abstract async exists() → bool[source]

Check if the table exists in the database.

Returns:

True if the table is present in the database, otherwise False.

Return type:

bool

abstract async create(if_not_exists: bool = True) → None[source]

Create the SQL table in the database.

Parameters:

if_not_exists (bool, optional) – If True, the table will be created if it does not already exist without raising an exception. Defaults to True.

abstract async drop(if_exists: bool = True) → None[source]

Drop the SQL table from the database.

Parameters:

if_exists (bool, optional) – If True, the table will be dropped only if it already exists without raising an exception. Defaults to True.

class sl3aio.table.SolidTable(name: str, _columns: tuple[TableColumn, ...], _executor: ConnectionManager)[source]

Bases: SqlTable, Generic

A concrete implementation of SqlTable for interacting with SQLite databases.

This class provides methods for performing CRUD (Create, Read, Update, Delete) operations on SQLite tables. It implements the abstract methods defined in SqlTable and Table classes.

Warning

You must call SolidTable.start_executor() or enter table’s async context using async with table construction before acessing the table, otherwise the program will await for the request to complete forever.

See also

TableColumn, TableRecord, SqlTable, Table

async __aenter__() → Self

Asynchronous context manager entry point.

Entries the executor’s context manager, allowing to interact with the table.

Returns:

The Table instance.

Return type:

Self

See also

ConsistentExecutor.__aenter__()

async __aexit__(*args) → None

Asynchronous context manager exit point.

Exits the executor’s context manager and disables the table.

Parameters:

*args – Arguments passed to the exit method.

See also

ConsistentExecutor.__aexit__()

property columns: tuple[TableColumn, ...]

The columns of the table.

async count(predicate: TableSelectionPredicate | None = None) → int

Count the number of records in the table that match the predicate.

Note

If no predicate is specified, the result will be the same as for the Table.length() method.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records. Defaults to None.

Returns:

Number of records that match the predicate.

Return type:

int

property database: str

Get the name of the database this table belongs to.

async delete(predicate: TableSelectionPredicate | None = None) → None

Delete records from the table.

Note

If predicate isn’t specified, clears the table.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records to delete. Defaults to None.

See also

Table.deleted(), Table.delete_one()

async delete_one(predicate: TableSelectionPredicate | None = None) → TableRecord | None

Delete a single record from the table.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the record to delete. Defaults to None.

Returns:

The deleted record, or None if no record matches.

Return type:

TableRecord [T] | None

See also

Table.delete(), Table.delete_one()

async classmethod from_database(name: str, executor: ConnectionManager) → SqlTable

Create a SqlTable instance from an existing database table.

Warning

Before awaiting this function, make sure that the executor is running, otherwise the program will freeze.

Parameters:

  • name (str) – The name of the existing table in the database.

  • executor (ConnectionManager) – The connection manager for the database.

Returns:

A new SqlTable instance representing the existing database table.

Return type:

SqlTable [T]

Raises:

AssertionError – If the table does not exist in the specified database.

async insert_many(ignore_existing: bool = False, *values: dict[str, T]) → AsyncIterator[TableRecord]

Insert multiple records into the table.

Important

You must iterate over the result for operation to be performed.

Parameters:

  • ignore_existing (bool, optional) – Whether to ignore existing records. Defaults to False.

  • *values (dict [str, T]) – Dictionaries containing the values for each record to insert.

Yields:

TableRecord [T] – The inserted records.

See also

Table.insert()

async make_record(*args: T, **kwargs: T) → TableRecord

Create a new record for this table.

For more details, see TableRecord.make().

Parameters:

  • *args (T) – Positional arguments for the record values.

  • **kwargs (T) – Keyword arguments for the record values.

Returns:

A new record for this table.

Return type:

TableRecord [T]

name: str

The name of the table.

async select_one(predicate: TableSelectionPredicate | None = None) → TableRecord | None

Select a single record from the table.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records. Defaults to None.

Returns:

The selected record, or None if no record matches.

Return type:

TableRecord [T] | None

See also

Table.select()

async start_executor() → None

Start the table’s executor.

async stop_executor() → None

Stop the table’s executor.

async update(predicate: TableSelectionPredicate | None = None, **to_update: T) → None

Update records in the table without yielding the updated records.

Note

If predicate isn’t specified, upadates every record.

Parameters:

  • predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records to update. Defaults to None.

  • **to_update (T) – The values to update in the matching records.

See also

Table.updated(), Table.update_one()

async update_one(predicate: TableSelectionPredicate | None = None, **to_update: T) → TableRecord | None

Update a single record in the table.

Parameters:

  • predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the record to update. Defaults to None.

  • **to_update (T) – The values to update in the matching record.

Returns:

The updated record, or None if no record matches.

Return type:

TableRecord [T] | None

See also

Table.updated(), Table.update()

async length() → int[source]

Get the number of records in the table.

Returns:

The number of records in the table.

Return type:

int

async contains(record: TableRecord) → bool[source]

Check if a record exists in the table.

Parameters:

record (TableRecord [T]) – The record to check for.

Returns:

True if the record exists in the table, False otherwise.

Return type:

bool

async insert(ignore_existing: bool = False, **values: T) → TableRecord[source]

Insert a new record into the table.

Parameters:

  • ignore_existing (bool, optional) – Whether to ignore existing records. Defaults to False.

  • **values (T) – The values for the new record.

Returns:

The inserted record.

Return type:

TableRecord [T]

See also

Table.insert_many()

async select(predicate: TableSelectionPredicate | None = None) → AsyncIterator[TableRecord][source]

Select records from the table.

Note

If predicate isn’t specified, yields the whole table.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records. Defaults to None.

Yields:

TableRecord [T] – The selected records.

See also

Table.select_one()

async deleted(predicate: TableSelectionPredicate | None = None) → AsyncIterator[TableRecord][source]

Remove and yield deleted records from the table.

Note

If predicate isn’t specified, yields the whole table and then clears it.

Important

You must iterate over the result for operation to be performed.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records to remove. Defaults to None.

Yields:

TableRecord [T] – The removed records.

See also

Table.delete(), Table.delete_one()

async updated(predicate: TableSelectionPredicate | None = None, **to_update: T) → AsyncIterator[TableRecord][source]

Update records in the table and yield the updated records.

Note

If predicate isn’t specified, updates and yields every record.

Important

You must iterate over the result for operation to be performed.

Parameters:

  • predicate (TableSelectionPredicate [T] | None, optional) – A predicate to filter the records to update. Defaults to None.

  • **to_update (T) – The values to update in the matching records.

Yields:

TableRecord [T] – The updated records.

See also

Table.update(), Table.update_one()

async exists() → bool[source]

Check if the table exists in the database.

Returns:

True if the table is present in the database, otherwise False.

Return type:

bool

async create(if_not_exists: bool = True) → None[source]

Create the SQL table in the database.

Parameters:

if_not_exists (bool, optional) – If True, the table will be created if it does not already exist without raising an exception. Defaults to True.

async drop(if_exists: bool = True) → None[source]

Drop the SQL table from the database.

Parameters:

if_exists (bool, optional) – If True, the table will be dropped only if it already exists without raising an exception. Defaults to True.