easytable¶

Description¶

This module provides a high-level, user-friendly interface for working with database tables in the sl3aio library. It offers simplified abstractions for common database operations, making it easier to define, query, and manipulate database tables.

The module aims to provide a more intuitive and Pythonic way of interacting with database tables, reducing the complexity often associated with SQL operations.

Key Components¶

EasySelector: A powerful class for building complex database queries and selections.
EasyColumn: A simplified way to define table columns with various attributes.
EasyTable: A high-level representation of database tables with easy-to-use methods for common operations.

Other Components¶

default_selector(): A default selection function used by EasySelector.

Usage Examples¶

Theory of EasySelector:

# The selector works as follows:
# 0. At a start point, selector's value is equal to the TableRecord.
# 1. You build a selection functions chain using EasySelector methods. Those are:
#    - Getters ('getattr', 'getitem')
#    - Function calls (if current selected value is callable)
#    - Logical operators ('==', '<', 'not', 'int', etc.). If failed, selector
#      will return False. After failure, some selections won't be applied.
#    - Arithmetic operators ('+', '-', '*', 'abs', etc.)
#    - Binary operators ('>>', '<<', '^', etc.)
#    - Other function calls using 'pass_into' method.
# 2. Then, when needed, selector will iterate over the table and apply itself
# on each record of the table, filtering and getting specific information
# from them.

# Selecting an attribute and getting item from current value:
selector.<attr_name>
selector[<item_name>]
# so to select a column at a start do: selector.<column_name>

# Checking current value against logical operations:
# For and, or, in, is and is not operators use and_, or_, in_, is_ and is_not_ methods.
selector == 'desired_value'
selector > 10
selector.in_([...])
selector.is_not_(None)
# ... and so on.

# Doing maths with current value:
selector + 10
selector - 5
round(selector)
# ... and so on.

# Making binary operations with current value:
selector >> 10
selector << 5
selector ^ 10
#... and so on.

# Passing current value into a custom function:
# Note 'key_or_pos' parameter. With it you can you can specify exactly how
# the value will be passed into the function.
selector.pass_into(my_function, key_or_pos=..., *other_args, **other_kwargs)

# Calling current value:
selector(...)

# Also, almost every selector operator, that takes 2 values, can be applied to 2 selectors:
selector1 + selector2
selector1 == selector2
selector1.and_(selector2)
selector1 in selector2
# ... and so on. This will apply selector2 on the record, that was on the start of the
# selector1 and then apply operator on the current value of selector1 and result of the
# selector2.

Here is an example of how to use the EasySelector class:

from sl3aio import EasySelector

async def main():
    # ... some code that loads table into variable 'my_table'.

    selector = EasySelector(my_table)  # Create a selector for the table

    # Set up the selector. It will pass for the records that are:
    # 1. Has 'name' column value equal to 'FooBar' and 'age' column value in range from 18 to 37
    # 2. Has True 'is_cool' column value
    # 3. Has 'data' column value decodable by key 'pass12345'.
    setted_up_selector = (
        selector.name == 'FooBar' and
        selector.age in range(18, 37) or
        selector.is_cool or
        selector.data.decode_with_key('pass12345').is_(True)
    )

    # Use the selector to get records from the table
    selected_records = await setted_up_selector.select()

If you already have a declared table and you want to use it with EasyTable, you can do this:

from sl3aio import EasyTable

async def main():
    # ... some code that loads table into variable 'my_table'.
    foo_table = EasyTable(my_table)  # Pass it into the EasyTable constructor.

    # And perform some operations on the table using EasyTable methods.
    await foo_table.insert(...)
    await foo_table.update(...)
    await foo_table.select(...)
    # ... and so on.

If you don’t have a table and do not want to create it using classes of the table module, you can create a markup using EasyTable and convert it into a working table. Then you can use EasyTable according to the previous example.

from sl3aio import EasyColumn, EasyTable, MemoryTable, TableColumnValueGenerator, EasySelector
from random import randint
from asyncio import run


# Custom value generator for user IDs. See TableColumnValueGenerator.
@TableColumnValueGenerator.from_function('userID')
def __generate_user_id() -> int:
    return randint(1000000, 9999999)


# Define the structure of the Users table using EasyTable
# EasyTable[int | str] indicates that column values can be either int or str
class UsersTableMarkup(EasyTable[int | str]):
    # Define columns with their types, properties, and constraints
    # EasySelector[int] specifies that this column will contain integer values
    id: EasySelector[int] = EasyColumn(TableColumnValueGenerator('userID'), primary=True, nullable=False)
    # 'John Doe' is set as the default value for the name column
    name: EasySelector[str] = 'John Doe'
    # age column is marked as non-nullable, requiring a value for every record
    age: EasySelector[int] = EasyColumn(nullable=False)
    # email column is defined without additional constraints
    email: EasySelector[str]


async def main():
    # Create a MemoryTable instance based on the UsersTableMarkup
    # This step converts the markup into actual table columns
    columns = UsersTableMarkup.columns()
    # Initialize an in-memory table named 'users' with the defined columns
    table = MemoryTable('users', columns)
    # Create a high-level interface for interacting with the users table
    users_table = UsersTableMarkup(table)

    # Insert a new user record into the table
    # The 'id' field is automatically generated using the custom generator
    await users_table.insert(name='Foo Bar', age=23, email='foobar@gmail.com')

    # Demonstrate querying capabilities
    # This line performs these operations:
    # 1. Create a query condition: users_table.name == 'Foo Bar'
    # 2. Select one record matching this condition: .select_one()
    # 3. Access the 'email' field of the selected record
    # 4. Print the result
    print((await (users_table.name == 'Foo Bar').select_one()).email)


run(main())  # >>> foobar@gmail.com

See also

table: Native module for working with tables.

sl3aio.easytable.default_selector(previous, _: TableRecord) → tuple[True, Any][source]¶

The default selector function for EasySelector that always returns True and the provided value.

Parameters:

previous (Any) – Value that was selected by selector previously.
_ (TableRecord) – The record that is being selected.

Returns:

Selection result. First value is an ‘ok’ flag, second value is a selection result.

Return type:

tuple [True, Any]

class sl3aio.easytable.EasySelector(table: ~sl3aio.table.Table | None = None, selector: ~collections.abc.Callable[[~typing.Any, ~sl3aio.table.TableRecord], tuple[bool, ~typing.Any]] = <function default_selector>, _predicate: ~sl3aio.table.TableSelectionPredicate | None = None)[source]¶

A class for creating and manipulating selectors for database operations.

This class provides methods for building complex selection criteria and performing various database operations based on those criteria. Just like EasyTable, performing database operations using this class does not require mannually closing/opening the connection. See the examples to understand how to use this class.

See also

TableSelectionPredicate

table: Table | None¶: The database table to operate on. Optional, defaults to None.

selector: Callable[[Any, TableRecord], tuple[bool, Any]]¶: The selector function. Optional, defaults to default_selector()

Attention

Change this field only if you know what you are doing.

as_predicate() → TableSelectionPredicate[source]¶

Creates a predicate function based on the current selector.

Returns:: An async function that takes a record and returns a boolean.
Return type:: TableSelectionPredicate [T]

pin_table(table: Table) → Self[source]¶

Creates a new EasySelector with the specified table.

Parameters:: table (Table [T]) – The table to pin to the selector.
Returns:: A new EasySelector instance with the specified table.
Return type:: EasySelector [T]

apply(record: TableRecord) → tuple[bool, Any][source]¶

Applies the selector to a given record.

Parameters:: record (TableRecord [T]) – The record to apply the selector to.
Returns:: A tuple containing a boolean indicating if the selector matched, and the result of the selector application.
Return type:: tuple [bool, Any]

async select(table: Table | None = None) → AsyncIterator[TableRecord][source]¶

Selects records from the table based on the current selector.

Parameters:: table (Table [T] | None, optional) – The table to select from. If None, uses the pinned table. Defaults to None.
Returns:: An async iterator of selected records.
Return type:: AsyncIterator [TableRecord [T]]

See also

EasySelector.select_one()

async select_one(table: Table | None = None) → TableRecord | None[source]¶

Selects a single record from the table based on the current selector.

Parameters:: table (Table [T] | None, optional) – The table to select from. If None, uses the pinned table. Defaults to None.
Returns:: The selected record, or None if no record matches.
Return type:: TableRecord [T] | None

See also

EasySelector.select()

async deleted(table: Table | None = None) → AsyncIterator[TableRecord][source]¶

Selects and removes records from the table based on the current selector.

Parameters:: table (Table [T] | None, optional) – The table to delete from. If None, uses the pinned table. Defaults to None.
Returns:: An async iterator of deleted records.
Return type:: AsyncIterator [TableRecord [T]]

async delete(table: Table | None = None) → None[source]¶

Deletes all records from the table that match the current selector.

Parameters:: table (Table [T] | None, optional) – The table to delete from. If None, uses the pinned table. Defaults to None.

async delete_one(table: Table | None = None) → TableRecord | None[source]¶

Deletes a single record from the table that matches the current selector.

Parameters:: table (Table [T] | None, optional) – The table to delete from. If None, uses the pinned table. Defaults to None.
Returns:: The deleted record, or None if no record matches.
Return type:: TableRecord [T] | None

async updated(table: Table, **to_update: T) → AsyncIterator[TableRecord][source]¶

Updates records in the table that match the current selector.

Parameters:

table (Table [T]) – The table to update.
**to_update (T) – Keyword arguments specifying the fields to update and their new values.

Returns:

An async iterator of updated records.

Return type:

AsyncIterator [TableRecord [T]]

async update(table: Table, **to_update: T) → None[source]¶

Updates all records in the table that match the current selector.

Parameters:

table (Table [T]) – The table to update.
**to_update (T) – Keyword arguments specifying the fields to update and their new values.

async update_one(table: Table, **to_update: T) → TableRecord | None[source]¶

Updates a single record in the table that matches the current selector.

Parameters:

table (Table [T]) – The table to update.
**to_update (T) – Keyword arguments specifying the fields to update and their new values.

Returns:

The updated record, or None if no record matches.

Return type:

TableRecord [T] | None

append_selector(selector: Callable[[bool, Any, TableRecord], tuple[bool, Any]]) → Self[source]¶

Appends a new selector to the current selector chain.

Parameters:: selector (Callable [[bool, Any, TableRecord [T]], tuple [bool, Any]]) – The selector to append.
Returns:: A new EasySelector instance with the appended selector.
Return type:: EasySelector

pass_into(func: Callable[[Concatenate[Any, P]], Any], *args: P, key_or_pos: str | int = 0, **kwargs: P) → Self[source]¶

Passes the result of the current selector into a function.

Parameters:

func (Callable [Concatenate [Any, P], Any]) – The function to pass the result into.
key_or_pos (int | str, optional) – Argument name or position in the function’s signature. Defaults to 0.
*args (P.args) – Positional arguments to pass to the function.
**kwargs (P.kwargs) – Keyword arguments to pass to the function.

Returns:

A new EasySelector instance with the modified selector.

Return type:

EasySelector

set_ok(value: bool = True) → Self[source]¶

Sets the ‘ok’ status of the selector to a fixed value.

Parameters:: value (bool, optional) – The value to set for the ‘ok’ status. Defaults to True.
Returns:: A new EasySelector instance with the modified selector.
Return type:: EasySelector

is_(other: Any | Self) → Self[source]¶

Checks if the result of the current selector is a given object.

Parameters:: obj (Any | Self) – The object to check against.
Returns:: A new EasySelector instance with the modified selector.
Return type:: EasySelector

is_not_(other: Any | Self) → Self[source]¶

Checks if the result of the current selector is not a given object.

Parameters:: obj (Any | Self) – The object to check against.
Returns:: A new EasySelector instance with the modified selector.
Return type:: EasySelector

in_(container: Container | Self) → Self[source]¶

Checks if the result of the current selector is in a container.

Parameters:: container (Container | Self) – The container to check against.
Returns:: A new EasySelector instance with the modified selector.
Return type:: EasySelector

and_(other: Self) → Self[source]¶

Combines the current selector with another using logical AND.

Parameters:: other (Self) – The other selector to combine with.
Returns:: A new EasySelector instance representing the combined selector.
Return type:: EasySelector

or_(other: Self) → Self[source]¶

Combines the current selector with another using logical OR.

Parameters:: other (Self) – The other selector to combine with.
Returns:: A new EasySelector instance representing the combined selector.
Return type:: EasySelector

not_() → Self[source]¶

Negates the current selector.

Returns:: A new EasySelector instance representing the negated selector.
Return type:: EasySelector

class sl3aio.easytable.EasyColumn(default: T | TableColumnValueGenerator | None = None, primary: bool = False, unique: bool = False, nullable: bool = True)[source]¶

Represents an easy-to-use column definition for database tables.

This class provides a simplified way to define columns for database tables, including options for default values, primary key, uniqueness, and nullability.

default: T | TableColumnValueGenerator | None¶: The default value for the column. Can be a static value, a TableColumnValueGenerator, or None. Defaults to None.

primary: bool¶: Indicates if this column is a primary key. Defaults to False.

unique: bool¶: Indicates if this column should have unique values. Defaults to False.

nullable: bool¶: Indicates if this column can contain NULL values. Defaults to True.

to_column(name: str, _type: type[T]) → TableColumn[source]¶

Converts the EasyColumn instance to a TableColumn instance.

This method creates a TableColumn object based on the EasyColumn’s attributes and the provided name and type.

Note

If a Parser wasn’t found for the given type, 'TEXT' is used as the default type.

Parameters:

name (str) – The name of the column.
_type (type [T]) – The Python type of the column’s values.

Returns:

A TableColumn instance representing the column in the database.

Return type:

TableColumn [T]

class sl3aio.easytable.EasyTable(table: Table)[source]¶

A class representing an easy-to-use selection interface for database tables.

This class provides methods for common database operations such as inserting, selecting, updating, and deleting records without having to open/close a connection manually. See the examples for more details.

table: Table¶: The underlying database table.

classmethod columns() → tuple[TableColumn, ...][source]¶

Get the columns of the table from the fields of the subclass.

Returns:: A tuple of TableColumn objects representing the table’s columns.
Return type:: tuple [TableColumn [T], …]

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 the table contains a specific record.

Parameters:: record (TableRecord [T]) – The record to check for.
Returns:: True if the record is 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) – If True, ignore if the record already exists. Defaults to False.
**values (T) – The values to insert, specified as keyword arguments.

Returns:

The inserted record.

Return type:

TableRecord [T]

See also

EasyTable.insert_many()

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

Insert multiple records into the table.

Parameters:

ignore_existing (bool, optinal) – If True, ignore if the records already exist. Defaults to False.
**values (dict [str, T]) – The values to insert, specified as dictionaries.

Returns:

An async iterator of the inserted records.

Return type:

AsyncIterator [TableRecord [T]]

See also

EasyTable.insert()

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

Select records from the table based on a predicate.

Parameters:: predicate (TableSelectionPredicate [T] | None, optional) – The selection predicate. Defaults to None.
Returns:: An async iterator of the selected records.
Return type:: AsyncIterator [TableRecord [T]]

See also

EasyTable.select_one()

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

Select a single record from the table based on a predicate.

Parameters:: predicate (TableSelectionPredicate [T] | None, optional) – The selection predicate. Defaults to None.
Returns:: The selected record, or None if no record matches the predicate.
Return type:: TableRecord [T] | None

See also

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

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

Remove and yield deleted records from the table based on a predicate.

Parameters:: predicate (TableSelectionPredicate [T] | None, optional) – The selection predicate. Defaults to None.
Returns:: An async iterator of the removed records.
Return type:: AsyncIterator [TableRecord [T]]

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

Delete records from the table based on a predicate.

Parameters:: predicate (TableSelectionPredicate [T] | None, optional) – The selection predicate. Defaults to None.

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

Delete a single record from the table based on a predicate.

Parameters:: predicate (TableSelectionPredicate [T] | None, optional) – The selection predicate. Defaults to None.
Returns:: The deleted record, or None if no record matches the predicate.
Return type:: TableRecord [T] | None

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

Update records in the table based on a predicate and return the updated records.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – The selection predicate. Defaults to None.
**to_update (T) – The values to update, specified as keyword arguments.

Returns:

An async iterator of the updated records.

Return type:

AsyncIterator [TableRecord [T]]

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

Update records in the table based on a predicate.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – The selection predicate. Defaults to None.
**to_update (T) – The values to update, specified as keyword arguments.

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

Update a single record in the table based on a predicate.

Parameters:

predicate (TableSelectionPredicate [T] | None, optional) – The selection predicate. Defaults to None.
**to_update (T) – The values to update, specified as keyword arguments.

Returns:

The updated record, or None if no record matches the predicate.

Return type:

TableRecord [T], None