executor¶

Description¶

This module provides asynchronous wrappers and utilities for working with SQLite databases in Python. It offers a set of classes that enable efficient, thread-safe, and consistent execution of SQLite operations within an asynchronous context.

The module is designed to work seamlessly with Python’s asyncio framework, providing an intuitive API for database operations in asynchronous applications.

Note

This module is designed for use with SQLite databases and may not be suitable for other database systems.
All database operations are executed in a consistent order, which may impact performance in some scenarios.
The module leverages Python’s asyncio framework, so it should be used within an asynchronous context.

Key Components¶

ConsistentExecutor: Ensures consistent order of execution for queued tasks.
ConnectionManager: Handles SQLite database connections asynchronously.
CursorManager: Manages SQLite cursor operations asynchronously.

Other Components¶

Executor: Base class for asynchronous execution of synchronous functions.
Connector: Manages SQLite connection parameters.
Parameters: Type alias for the allowed SQL parameters.

Usage Examples¶

Basic Usage with In-Memory Database:

import asyncio
from sl3aio.executor import ConnectionManager, Connector

async def main():
    connector = Connector(":memory:")
    async with ConnectionManager(connector) as cm:
        # Create a table
        await cm.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")

        # Insert data
        await cm.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))
        await cm.execute("INSERT INTO users (name) VALUES (?)", ("Bob",))

        # Query data
        cursor = await cm.execute("SELECT * FROM users")
        users = await cursor.fetch()

        for user in users:
            print(f"User: {user}")

asyncio.run(main())

Using CursorManager for Iteration:

import asyncio
from sl3aio.executor import ConnectionManager, Connector

async def main():
    connector = Connector(":memory:")
    async with ConnectionManager(connector) as cm:
        await cm.execute("CREATE TABLE numbers (n INTEGER)")
        await cm.executemany("INSERT INTO numbers VALUES (?)", [(i,) for i in range(10)])

        cursor = await cm.execute("SELECT * FROM numbers")
        async for row in cursor:
            print(f"Number: {row[0]}")

asyncio.run(main())

Transaction Management:

import asyncio
from sl3aio.executor import ConnectionManager, Connector

async def main():
    connector = Connector(":memory:")
    async with ConnectionManager(connector) as cm:
        await cm.execute("CREATE TABLE accounts (id INTEGER PRIMARY KEY, balance REAL)")

        try:
            await cm.execute("INSERT INTO accounts (balance) VALUES (?)", (100,))
            await cm.execute("UPDATE accounts SET balance = balance - ? WHERE id = ?", (50, 1))
            await cm.execute("UPDATE accounts SET balance = balance + ? WHERE id = ?", (50, 2))
            await cm.commit()
            print("Transaction committed successfully")
        except Exception as e:
            await cm.rollback()
            print(f"Transaction rolled back: {e}")

asyncio.run(main())

Using ConsistentExecutor for Ordered Task Execution:

import asyncio
from sl3aio.executor import ConsistentExecutor

async def main():
    async with ConsistentExecutor() as executor:
        def task(n):
            print(f"Executing task {n}")
            return f"Result {n}"

        futures = [executor(task, i) for i in range(5)]
        results = await asyncio.gather(*futures)
        print("Results:", results)

asyncio.run(main())

sl3aio.executor.Parameters: TypeAlias = collections.abc.Sequence[bytes | str | int | float | None] | collections.abc.Mapping[str, bytes | str | int | float | None]¶: Allowed SQL request parameters type.

class sl3aio.executor.Executor[source]¶

Bases: object

A class that provides asynchronous execution capabilities for synchronous functions.

This class uses a ThreadPoolExecutor to run synchronous functions in a separate thread, allowing them to be executed asynchronously without blocking the main event loop.

Example

import time
from asyncio import run

executor = Executor()

def slow_function(duration):
    time.sleep(duration)
    return f"Slept for {duration} seconds"

async def main():
    result = await executor(slow_function, 2)
    print(result)

run(main())

# Output:
# Slept for 2 seconds

Notes

The Executor class is designed to work with Python’s asyncio framework.
It automatically uses the running event loop and creates a new ThreadPoolExecutor.
This class is useful for running CPU-bound or blocking I/O operations without blocking the main event loop.

__call__(func: Callable[[P], R], *args: P, **kwargs: P) → Future[source]¶

Execute the given function asynchronously in a separate thread.

This method allows you to run any synchronous function asynchronously, which is particularly useful for CPU-bound tasks or operations that would otherwise block the event loop.

Parameters:

func (Callable [P, R]) – The function to be executed asynchronously.
*args (P.args) – Positional arguments to be passed to the function.
**kwargs (P.kwargs) – Keyword arguments to be passed to the function.

Returns:

A Future object representing the eventual result of the function call.

Return type:

Future [R]

class sl3aio.executor.ConsistentExecutor[source]¶

Bases: Executor

A class that provides consistent asynchronous execution capabilities for synchronous functions.

This class extends the Executor class and ensures that tasks are executed in a consistent order using a single-threaded executor and a queue system. It’s particularly useful when you need to maintain the order of execution for a series of tasks.

Example

import asyncio
from sl3aio.executor import ConsistentExecutor

async def main():
    async with ConsistentExecutor() as executor:
        # Define some example functions
        def task1():
            print("Executing task 1")
            return "Result 1"

        def task2():
            print("Executing task 2")
            return "Result 2"

        # Queue the tasks
        future1 = executor(task1)
        future2 = executor(task2)

        # Wait for the results
        result1 = await future1
        result2 = await future2

        print(f"Results: {result1}, {result2}")

asyncio.run(main())

# Output:
# Executing task 1
# Executing task 2
# Results: Result 1, Result 2

Notes

The ConsistentExecutor ensures that tasks are executed in the order they are queued.
It uses a single thread for execution, which can be beneficial for maintaining consistency but may not be suitable for CPU-bound tasks that require parallelism.
The class implements the async context manager protocol, allowing for easy resource management.

property running: bool¶: Check if the worker task is currently running.

async start() → bool[source]¶

Start the worker task if it’s not already running.

This method increments the reference count and creates a new worker task if one doesn’t exist.

Returns:: True if a new worker task was created, False otherwise.
Return type:: bool

async stop() → bool[source]¶

Stop the worker task if there are no more references.

This method decrements the reference count and, if it reaches zero, waits for all queued tasks to complete before cancelling the worker task.

Returns:: True if the worker task was stopped, False otherwise.
Return type:: bool

__call__(func: Callable[[P], R], *args: P, **kwargs: P) → Future[source]¶

Queue a function for execution and return a Future.

This method wraps the given function and its arguments into a partial function, creates a new Future, and adds both to the queue for later execution.

Parameters:

func (Callable [P, R]) – The function to be executed.
*args (P.args) – Positional arguments for the function.
**kwargs (P.kwargs) – Keyword arguments for the function.

Returns:

A Future representing the eventual result of the function call.

Return type:

Future [R]

async __aenter__() → Self[source]¶

Async context manager entry point.

This method is called when entering an ‘async with’ block. It starts the worker task.

Returns:: The ConsistentExecutor instance.
Return type:: Self

async __aexit__(*args) → None[source]¶

Async context manager exit point.

This method is called when exiting an ‘async with’ block. It stops the worker task and raises any exceptions that occurred during execution.

Parameters:: *args (Any) – Exception details (type, value, traceback) if an exception occurred.

class sl3aio.executor.Connector(dbfile: dataclasses.InitVar[Union[str, bytes, pathlib.Path, Literal[':memory:']]], timeout: float = 5.0, detect_types: int = 0, isolation_level: Literal['DEFERRED', 'EXCLUSIVE', 'IMMEDIATE'] | None = 'DEFERRED', check_same_thread: bool = False, factory: type | None = None, cached_statements: int = 128, uri: bool = False, autocommit: bool = False)[source]¶

Bases: object

This class allows to save and reuse parameters of the sqlite3.connect method. Every attribute of this class is similar to the original parameters.

During initialization, it normalizes the database path and ensures that it exists.

Example

# Create a Connector instance
connector = Connector(database="my_database.db", timeout=10.0, autocommit=True)

# Establish a connection to the database
connection = connector()

dbfile: dataclasses.InitVar[Union[str, bytes, pathlib.Path, Literal[':memory:']]]¶

‘ will be interpreted as a path.

Note

This is InitVar, so after initialization use Connector.database attribute instead.

Type:: Specifies the path to the database file. Every value except the ‘
Type:: memory

timeout: float¶

detect_types: int¶

isolation_level: Literal['DEFERRED', 'EXCLUSIVE', 'IMMEDIATE'] | None¶

check_same_thread: bool¶

factory: type | None¶

cached_statements: int¶

uri: bool¶

autocommit: bool¶

database: Path | Literal[':memory:']¶

connect() → Connection[source]¶

This method establishes and returns a connection to the database with specified parameters.

Returns:: Connection object.
Return type:: Connection

connection_manager() → ConnectionManager[source]¶

This method returns a ConnectionManager instance with pre-defined connector parameter.

Returns:: An instance of the connection manager associated with the current connector.
Return type:: ConnectionManager

class sl3aio.executor.ConnectionManager(connector: Connector)[source]¶

Bases: ConsistentExecutor

A class that manages SQLite database connections asynchronously.

This class extends the ConsistentExecutor class and provides a consistent, thread-safe way to interact with SQLite databases. It ensures that database operations are executed in the order they are queued, using a single connection per database.

Example

import asyncio
from sl3aio.executor import ConnectionManager, Connector

async def main():
    # Create a Connector for an in-memory database
    connector = Connector(":memory:")

    # Create a ConnectionManager
    async with ConnectionManager(connector) as cm:
        # Create a table
        await cm.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")

        # Insert some data
        await cm.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))
        await cm.execute("INSERT INTO users (name) VALUES (?)", ("Bob",))

        # Query the data
        cursor = await cm.execute("SELECT * FROM users")
        users = await cursor.fetch()

        for user in users:
            print(f"User: {user}")

asyncio.run(main())

Notes

The ConnectionManager uses a singleton pattern to ensure only one instance per database file.
All database operations are executed in a consistent order using the underlying ConsistentExecutor.
The class implements the async context manager protocol for easy resource management.