functions - SQL functions

class SQLFunctionsRegistry[source]

Bases: ClassRegistry

Registry for SQL functions.

__init__()
can_register(obj) → bool

Check if an object can be registered.

clear() → None

Unlink all registered objects. Use it with caution.

find(condition: Callable[[Any], bool]) → _Obj

Find an object matching a condition.

find_all(condition: Callable[[Any], bool]) → Generator[_Obj, None, None]

Find all objects matching a condition.

find_subclass(bases: Collection[_Obj] | _Obj) → _Obj

Find a first subclass matching bases. A shortcut to find method.

find_subclasses(bases: Collection[_Obj] | _Obj) → Generator[_Obj, None, None]

Find all subclasses matching bases. A shortcut to find_all method.

get(k[, d]) → D[k] if k in D, else d.  d defaults to None.
register(obj: _Obj, name: _Key = None) → _Key

Register an object in the registry and return a key under which it has been registered.

Parameters:

  • obj – object to register

  • name – provide a custom name (not recommended)

Raises:

RegistrationFailed – if an object can’t be registered

Returns:

object key in the registry

register_from_module(module: object, *, ignore_key_names: bool = True) → FrozenSet[_Key]

Register classes from current object.

Parameters:

  • module – any object with __dict__

  • ignore_key_names – set it to True to ignore namespace keys when settings object names

Returns:

a set of registered keys

register_from_namespace(namespace: Mapping, *, ignore_key_names: bool = True) → FrozenSet[_Key]

Register all supported objects from an arbitrary mapping.

Incompatible objects will be ignored. Returns a set of registered keys.

Parameters:

  • namespace – any mapping

  • ignore_key_names – set it to True to ignore namespace keys when settings object names

Returns:

a set of registered keys

register_many(obj: Collection[_Obj]) → Tuple[_Key, ...]

Register multiple objects at once.

Parameters:

obj – objects

Raises:

RegistrationFailed – if any of the objects can’t be registered

Returns:

a tuple of object keys

SQL_FUNCTIONS = SQLFunctionsRegistry(objects={}, raise_if_exists=False)

sql functions registry

class UserFunction[source]

Function interface for SQL functions.

class MyFunction(UserFunction):
    package = 'utils'
    name = 'my_func'
    type = sa.types.Integer
    body = "my_func(v integer) RETURNS integer AS $$ BEGIN RETURN v; END; $$ LANGUAGE PLPGSQL;"

    def __init__(self, v: int):
        return super().__init__(v)

Now you can use this function from python. See https://docs.sqlalchemy.org/en/13/core/functions.html for detail about generic functions methods.

await self.app.db.fetchval(MyFunction(42).select())

To make your function actually available in your app you need to tell the database service about it, i.e. register it in the function registry

from kaiju_db.services import functions_registry

functions_registry.register_class(UserFunction)

or (if you have many)

from kaiju_db.services import functions_registry

import my_functions

functions_registry.register_from_module(my_functions)
__init__(*args: _ColumnExpressionOrLiteralArgument[Any], **kwargs: Any)

Construct a Function.

The func construct is normally used to construct new Function instances.

class DDL[source]

Specifies literal SQL DDL to be executed by the database. DDL objects function as DDL event listeners, and can be subscribed to those events listed in DDLEvents, using either Table or MetaData objects as targets. Basic templating support allows a single DDL instance to handle repetitive tasks for multiple tables.

example create function before create table task:

DDL(
    task,
    "before_create",
    """
        CREATE OR REPLACE FUNCTION TaskSortingIncrement()
        RETURNS trigger AS $BODY$BEGIN
           NEW.sort:= (SELECT COALESCE(max(sort), 0) FROM task WHERE status_id = NEW.status_id) + 1;
           RETURN NEW;
        END
        $BODY$
        LANGUAGE plpgsql VOLATILE
        COST 100;
    """
)

example create trigger after create table task:

DDL(
    task,
    "after_create",
    """
        DROP TRIGGER IF EXISTS TaskTriggerBeforeInsert
        ON task;
        CREATE TRIGGER TaskTriggerBeforeInsert
        BEFORE INSERT
        ON task
        FOR EACH ROW
        EXECUTE PROCEDURE TaskSortingIncrement();
    """
)
Parameters:
__init__()