BDC-DB Extension API

Database management extension for Brazil Data Cube applications and services.

class bdc_db.ext.BrazilDataCubeDB(app=None, **kwargs)

Database management extension for Brazil Data Cube applications and services.

alembic: A Flask-Alembic instance used to prepare migration environment.

init_app(app, **kwargs)

Initialize Flask application instance.

This method prepares the Alembic configuration for multiple named branches according to each package entry point.

Parameters:

app – Flask application
kwargs – Optional arguments to Flask-SQLAlchemy.

Keyword Arguments:

entry_point_group (str) – Custom entry point group for SQLAlchemy database models.
entry_point_jsonschemas (str) – Custom entry point group for JSONSchemas
engine_options (dict) – Custom SQLAlchemy Engine Options for instance object.

init_db(app, entry_point_group: str = 'bdc_db.models', engine_options=None, **kwargs)

Initialize Flask-SQLAlchemy extension.

Parameters:

app – Flask application
entry_point_group – Entrypoint definition to load models
engine_options – DB instance engine options
kwargs – optional Arguments to Flask-SQLAlchemy.

load_namespaces(entry_point: str = 'bdc_db.namespaces')

Load application namespaces dynamically using entry points.

Parameters:: points. (entry_point - Pattern to search in the setup.py entry) –

load_scripts(entry_point_group: str = 'bdc_db.scripts', **kwargs): Load SQL files from packages to BDC-DB context.

load_triggers(entry_point_group: str = 'bdc_db.triggers', **kwargs)

Load trigger files from packages to BDC-DB context.

Seeks for .sql files in packages which set bdc_db.triggers entry point.

Notes

It may throw exception when module is set, but does not exists in disk.

Parameters:: points. (entry_point_group - Pattern to search in the setup.py entry) –

register_scripts(module_name: str, trigger_name: str, path: str): Register trigger command to BDC-DB.

register_trigger(module_name: str, trigger_name: str, path: str): Register trigger command to BDC-DB.

bdc_db.ext.alembic_include_object(object, name, type_, reflected, compare_to)

Ignores the tables in ‘exclude_tables’.

For more information, please, refer to the Runtime Objects section in the Alembic documentation.

Parameters:

object – A SchemaItem object ( Table, Column, Index UniqueConstraint, or ForeignKeyConstraint object).
name – The name of the object.
type – A string describing the type of object (“table”, “column”, “index”, “unique_constraint”, or “foreign_key_constraint”).
reflected – True if the given object was produced based on table reflection, False if it’s from a local MetaData object.
compare_to – The object being compared against, if available, else None.

Returns:

True if the given object should be considered in the autogenerate sweep, otherwise, returns False.

Database instance using Flask-SQLAlchemy extension.

bdc_db.db.NAMING_CONVENTION = {'ck': '%(table_name)s_%(constraint_name)s_ckey', 'fk': '%(table_name)s_%(column_0_name)s_%(referred_table_name)s_fkey', 'ix': 'idx_%(column_0_label)s', 'pk': '%(table_name)s_pkey', 'uq': '%(table_name)s_%(column_0_name)s_key'}: Naming convention for SQLAlchemy constraint keys

bdc_db.db.db = <SQLAlchemyDB>: Shared database instance using Flask-SQLAlchemy extension.

bdc_db.db.metadata = MetaData(): Default database metadata object holding associated schema constructs.

SQL Types

Represent the custom data types for BDC-Catalog.

class bdc_db.sqltypes.JSONB(schema: str, draft_checker=None, *args, **kwargs)

Represent a Custom Data Type for dealing with JSONB and JSONSchemas on SQLAlchemy.

The bdc_db.sqltypes.JSONB type includes the fully support of sqlalchemy.dialects.postgresql.JSONB, including the JSONSchema validator.

New in version 0.6.0.

Examples

This examples describes a minimal way to use bdc_db.sqltypes.JSONB. It expects you have module structure like described in Usage, and the myapp/myschema.json is defined as:

{
    "$schema": "http://json-schema.org/draft-07/schema",
    "$id": "myapp.json",
    "type": "object",
    "title": "Teste",
    "required": [
        "mykey"
    ],
    "properties": {
        "mykey": {
            "type": "number",
            "minimum": 0,
            "maximum": 100
        }
    }
}

In the JSONSchema above, it contains the key mykey defined, which has the constraint that the value MUST be between 0 and 100. The following section describes a minimal way to insert your model into database:

>>> from bdc_db.db import db
>>> from bdc_db.sqltypes import JSONB
>>> class Collection(db.Model):
...      # Define a simple table to store collections.
...      __table_name__ = 'collections'
...      id = db.Column(db.Integer, primary_key=True, autoincrement=True)
...      name = db.Column(db.String, nullable=False)
...      properties = db.Column(JSONB('myapp/myschema.json'))
>>> c = Collection()
>>> c.name = 'Teste'
>>> c.properties = {"mykey": 10}
>>> db.session.add(c)
>>> db.session.commit()
>>> c.properties = {"mykey": 102}
>>> db.session.commit()  # Error

Utils

Utility for Image Catalog Extension.

class bdc_db.utils.TriggerResult(schema: str, table_name: str, trigger_schema: str, trigger_name: str, event: str, definition: str): Represent a Queryable Trigger Result.

bdc_db.utils.delete_trigger(name: str, engine: Engine, table: str, schema: str = None)

Delete a trigger context (if exists) on database.

Parameters:

name (str) – The trigger name.
engine (Engine) – The SQLAlchemy active database engine.
table (str) – The table name.
schema (str) – The table schema that the trigger is attached.

bdc_db.utils.execute(statement: str | Any, executor: Engine | Any, *args, **kwargs)

Execute a query statement in SQLAlchemy database engine.

Parameters:

statement – Query or SQLAlchemy query expression to execute
executor – The database engine to create connections

bdc_db.utils.has_schema(engine: Engine, schema: str, **kwargs) → bool

Check if the database schema (namespace) exists.

Parameters:

engine – The SQLAlchemy engine object.
schema – Database schema (namespace).

Keyword Arguments:

dialect. (** Any extra parameter supported by Engine) –

bdc_db.utils.list_triggers(engine: Engine) → List[TriggerResult]

List all the available triggers on current engine.

Parameters:: engine (Engine) – The activate SQLAlchemy database connector.

bdc_db.utils.validate_schema(schema_key: str, value: Any, draft_checker=None) → Any

Validate a JSONSchema according a json model.

New in version 0.6.0.

Raises:: jsonschema.ValidationError – When the current value does not match with expected schema.

Note

This method works under a Flask Application Context since it seeks for JSONSchema loaded into BrazilDataCubeDB()

Parameters:

schema_key (str) – The schema key path reference to the jsonschemas folder.
value (Any) – The model value to be validated.
draft_checker (jsonschema.FormatChecker) – The format checker validation for schemas.