functions - SQL functions

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__()