💧 Mayim

# mayim.mayim

# Table of Contents

# Mayim

class Mayim()

[view_source] (opens new window)

Main entryway for initializing access to the data layer.

Where possible, it is advised (although not necessary) to initialize the Mayim object once the async loop is running.

Example:

async def run():
    Mayim(
        executors=[MyExecutor]
        dsn="postgres://user:pass@localhost:5432/mydb"
    )

# __init__

def __init__(*,
             executors: Optional[Sequence[Union[Type[Executor],
                                                Executor]]] = None,
             dsn: str = "",
             db_path: str = "",
             hydrator: Optional[Hydrator] = None,
             pool: Optional[BaseInterface] = None,
             strict: bool = True)

[view_source] (opens new window)

Initializer for Mayim instance

The dsn, db_path, and the pool are mutually exclusive. Either one or none of them should be passed. If one exists, it will be used to create the fallback interface pool for the datasource. Therefore, if an executor does not define its own, then it will use the fallback pool as established here.

Similarly, the hydrator is used in the case that an executor does not define a hydrator. If one is not provided, then a generic hydrator instance will be created.

Executors need not be registered when the Mayim instance is instantiated. See load and loading and instantiating for more details.

Arguments:

  • executors Sequence[Union[Type[Executor], Executor]], optional - Executors that are being loaded at initialization time. Defaults to None.
  • dsn str, optional - DSN to the data source. Defaults to "".
  • db_path str, optional - Path to a SQLite database. Defaults to "".
  • hydrator Hydrator, optional - Fallback hydrator to use if not specified. Defaults to None.
  • pool BaseInterface, optional - Fallback database interface. Defaults to None.
  • strict bool, optional - Whether to raise an error if there is an empty method in a decorator but no loaded query. Defaults to True.

Raises:

  • MayimError - If there is conflicing data access source

# connect

async def connect() -> None

[view_source] (opens new window)

Connect to all database interfaces

# disconnect

async def disconnect() -> None

[view_source] (opens new window)

Disconnect from all database interfaces

# get

@staticmethod
def get(executor: Type[T]) -> T

[view_source] (opens new window)

Fetch an instance of an executor

This is useful when there you wish to retrieve the registered instance of an executor. Can be used in various parts of an application without having to pass variables.

Example:

from mayim import Mayim
from my.package.executors import AwesomeExecutor

async def some_func():
    executor = Mayim.get(AwesomeExecutor)
    awesomeness = await executor.select_something_awesome()
    ...

Arguments:

  • executor Type[T] - The class of the registered executor instance

Raises:

  • MayimError - If the passed executor has not been registered

Returns:

  • Executor - The executor instance

# load

def load(*,
         executors: Sequence[Union[Type[Executor], Executor]],
         hydrator: Optional[Hydrator] = None,
         pool: Optional[BaseInterface] = None,
         strict: Optional[bool] = None) -> None

[view_source] (opens new window)

Look through the Executor definition for methods that should execute a query, then load the corresponding SQL from a .sql file and hold in memory.

Arguments:

  • executors Sequence[Union[Type[Executor], Executor]] - The executor classes or instances to be registered
  • hydrator Hydrator, optional - Fallback hydrator to use if not specified. Defaults to None.
  • pool BaseInterface, optional - Fallback database interface. Defaults to None.
  • strict bool, optional - Whether to raise an error if there is an empty method in a decorator but no loaded query. Defaults to True.

Raises:

  • MayimError - If there is no fallback database interface, Mayim will raise an exception if there is a registered executor without an instance-specific database interface

# register

def register(cls, executor: Executor) -> None

[view_source] (opens new window)

Register an Executor instance

Arguments:

  • executor Executor - The instance to be registered

# reset_registry

def reset_registry()

[view_source] (opens new window)

Empty the executor registry

# transaction

@classmethod
@asynccontextmanager
async def transaction(cls, *executors: Union[SQLExecutor, Type[SQLExecutor]])

[view_source] (opens new window)

← mayim.registry →