💧 Mayim

# mayim.base.executor

# Table of Contents

# Executor

class Executor(Generic[T])

[view_source] (opens new window)

Base class for creating executors, which serve as the main interface for interacting with a data source. Likely you will want to create a subclass from one of its subclasses and not directly from this base class.

# __init__

def __init__(pool: Optional[BaseInterface] = None,
             hydrator: Optional[Hydrator] = None) -> None

[view_source] (opens new window)

Base class for creating executors

Arguments:

  • pool BaseInterface, optional - An interface used for a specific executor to override a global pool. Defaults to None.
  • hydrator Hydrator, optional - A hydrator used for a specific executor to override a global hydrator. Defaults to None.

Raises:

  • MayimError - If a dependency is missing

# execute

def execute(query: Union[str, Query],
            name: str = "",
            model: Optional[Type[object]] = None,
            as_list: bool = False,
            allow_none: bool = False,
            posargs: Optional[Sequence[Any]] = None,
            params: Optional[Dict[str, Any]] = None)

[view_source] (opens new window)

Low-level API to execute a query and hydrate the results

Arguments:

  • query Union[str, Query] - The query to be executed
  • name str, optional - The name of the query. Defaults to "".
  • model Type[object], optional - The model to be used for hydration. Defaults to None.
  • as_list bool, optional - Whether to return the results as a list of hydrated objects. Defaults to False.
  • allow_none bool, optional - Whether None is an acceptable return value. Defaults to False.
  • posargs Sequence[Any], optional - Positional arguments. Defaults to None.
  • params Dict[str, Any], optional - Keyword arguments. Defaults to None.

# get_base_path

@classmethod
def get_base_path(cls, directory_name: Optional[str]) -> Path

[view_source] (opens new window)

Get the base path for where queries will be located

Arguments:

  • directory_name str, optional - A starting directory

Raises:

  • MayimError - When a module does not exist

Returns:

  • Path - The base path

# get_hydrator

def get_hydrator(name: Optional[str] = None) -> Hydrator

[view_source] (opens new window)

Return a hydrator

Arguments:

  • name str, optional - The name of the hydrator to be retrieved. If no name is supplied, it will look for the hydrator by the name of the calling method. Defaults to None.

Raises:

  • MayimError - When the hydrator coul not be found

Returns:

  • Hydrator - A hydrator object

# get_query

def get_query(name: Optional[str] = None) -> T

[view_source] (opens new window)

Return a query

Arguments:

  • name str, optional - The name of the query to be retrieved. If no name is supplied, it will look for the query by the name of the calling method. Defaults to None.

Raises:

  • MayimError - When the query coul not be found

Returns:

  • Query - A query object

# hydrator

@property
def hydrator() -> Hydrator

[view_source] (opens new window)

The assigned hydrator. Will return an instance specific hydrator if one was assigned.

Returns:

  • Hydrator - The hydrator

# is_query_name

@staticmethod
def is_query_name(obj) -> bool

[view_source] (opens new window)

Whether an object is a valid query name

Arguments:

  • obj Any - Depends upon which subclass is being implemented

Returns:

  • bool - is it valid

# path: Optional[Union[str, Path]]

Optional[Union[str, Path]]: Class property that is a custom path " location of queries to be loaded. Default to None

Default: None

# pool

@property
def pool() -> BaseInterface

[view_source] (opens new window)

The assigned pool. Will return an instance specific pool if one was assigned.

Returns:

  • BaseInterface - The pool interface

# run_sql

def run_sql(query: str = "",
            name: str = "",
            as_list: bool = False,
            no_result: bool = False,
            posargs: Optional[Sequence[Any]] = None,
            params: Optional[Dict[str, Any]] = None)

[view_source] (opens new window)

Low-level API to execute a query and return the results without hydration.

Arguments:

  • query Union[str, Query] - The query to be executed
  • name str, optional - The name of the query. Defaults to "".
  • as_list bool, optional - Whether to return the results as a list of hydrated objects. Defaults to False.
  • allow_none bool, optional - Whether None is an acceptable return value. Defaults to False.
  • posargs Sequence[Any], optional - Positional arguments. Defaults to None.
  • params Dict[str, Any], optional - Keyword arguments. Defaults to None.

# is_auto_exec

def is_auto_exec(func) -> bool

[view_source] (opens new window)

Check if a method should be auto-executed.

Example:

Any of the following are acceptable:

async def method_ellipsis(self) -> None:
    ...

async def method_pass(self) -> None:
    pass

async def method_docstring(self) -> None:
    '''This is a docstring'''

Arguments:

  • func - The function or method being checked

Returns:

  • bool - Whether the function is empty and should be auto-executed

← mayim.base.hydrator →