aiodesa.database module

aiodesa.Database: Simple SQLite Database Interface

This module provides the Db class, a simple SQLite database interface that supports asynchronous operations.

Classes:

Db: Represents a simple SQLite database interface.

Example:

from aiodesa import Db

class Users:
    username: str
    id: str | None = None
    table_name: str = "users"

async with Db("database.sqlite3") as db:
    await db.read_table_schemas(Users)

class aiodesa.database.Db(db_path: str)[source]

Bases: object

Represents a simple SQLite database interface.

Parameters:: db_path – str The path to the SQLite database file.

Example:

class Users:
        username: str
        id: str | None = None
        table_name: str = "users"

async with Db("database.sqlite3") as db:
    await db.read_table_schemas(Users)
    ...

db_path: Path

delete(data_class: Any, column_identifier: None | str = None) → Callable[[...], Coroutine[Any, Any, None]][source]

Create a record deletion operation for the specified table. This defaults to the primary key ifthe column_identifier is not provided.

Parameters:

data_class – The data class representing the table structure.
column_identifier – The column to use for identifying records.

Returns:

A function to be called with the identifier for record deletion.

Example:

class Users:
        username: str
        id: str | None = None
        table_name: str = "users"

async with Db("database.sqlite3") as db:
            await db.read_table_schemas(Users)

    ...

    delete = db.delete(UserEcon)
    await delete("john_doe")

find(data_class: Any, column_identifier: None | str = None) → Callable[[...], Coroutine[Any, Any, None]][source]

Create a record retrieval operation for the specified table.

Parameters:

data_class – The data class representing the table structure.
column_identifier – The column to use for identifying records.

Defaults to the primary key of the data class if not specified.

Returns:: A function to be called with the identifier for record retrieval.

Example:

class MyBestFriends:
        username: str
        id: str | None = None
        table_name: str = "users"

async with Db("database.sqlite3") as db:
            await db.read_table_schemas(MyBestFriends)

    ...

    find_jimmy = db.find(MyBestFriends)
    jimmy = await find_jimmy("jimmy")

find_all(data_class: Any) → Callable[[...], Coroutine[Any, Any, list]][source]

Create record retrieval operation to fetch all records from the specified table.

Parameters:: data_class – The data class representing the table structure.
Returns:: A function to be called with optional additional query parameters.

Example:

class Users:
        username: str
        id: str | None = None
        table_name: str = "users"

async with Db("database.sqlite3") as db:
    await db.read_table_schemas(Users)
    ...
    find_all_users = db.find_all(Users)
    all_users = await find_all_users()

insert(data_class: Any) → Callable[[...], Coroutine[Any, Any, None]][source]

Create a record and insert it into the specified table.

Parameters:: data_class – The data class representing the table structure.
Returns:: A function to be called with the record data.

Example:

class Users:
        username: str
        id: str | None = None
        table_name: str = "users"

async with Db("database.sqlite3") as db:
            await db.read_table_schemas(Users)
    ...
    insert = db.update(UserEcon)
    await insert("john_doe")

async read_table_schemas(class_obj: Any | Tuple[Any, ...]) → None[source]

Read table schemas and create tables in the database.

Parameters:: schema – The schema or tuple of schemas to be processed. Each schema should be a data class representing a table.
Returns:: This method does not return any value.

Example:

class Users:
        username: str
        id: str | None = None
        table_name: str = "users"

async with Db("database.sqlite3") as db:
            await db.read_table_schemas(Users)
    ...

Note

Provide any additional notes or considerations about the method.

update(data_class: Any, column_identifier: None | str = None) → Callable[[...], Coroutine[Any, Any, None]][source]

Create a record update operation for the specified table.

Parameters:

data_class – The data class representing the table structure.
column_identifier – The column to use for identifying records.

Returns:

A function to be called with the record data for updating.

Example:

class Users:
        username: str
        id: str | None = None
        table_name: str = "users"

async with Db("database.sqlite3") as db:
    await db.read_table_schemas(Users)
    ...
    update = db.update(UserEcon)
    await update("john_doe")

Note: If the column_identifier is not provided, the primary key of the data class will be used as the identifier.

Subpackages

aiodesa.database module