Controller

First, we'll go over the basics and examine the Controller class. Then, we'll finish up our example.

The Controller class

You can import the Controller like so:

from dbdeclare.controller import Controller

The Controller executes commands in the cluster. It looks at everything you've declared and (depending on the function you call) creates/drops entities and grants/revokes privileges. Under the hood, it iterates over all entities and executes their specific commands. So if all you have declared is something like:

db = Database(name="dev")

The only SQL executed from Controller.run_all will be:

CREATE DATABASE dev;

There are a handful of functions to choose from. run_all and remove_all are the most likely entrypoints. run_all runs create_all which creates all declared entities, then runs grant_all which grants all declared privileges. remove_all runs revoke_all which revokes all privileges, then runs drop_all which drops all declared entities. Take a look at the class docstrings for more detail. The Controller interacts heavily with the underlying Entity class and is to some extent a wrapper around it.

For what it's worth, this is where a lot of future development will go: we'd like to eventually have updates, change detection, integration with Alembic, and more.

Example

Let's finish up our example. We have all our entities declared, and we have all our grants declared as well. Now we just need to execute:

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

from dbdeclare.controller import Controller

# omitted code between

    # create engine with admin user and default database
    engine = create_engine(url="postgresql+psycopg://postgres:postgres@127.0.0.1:5432/postgres")
    # create all entities and grant all privileges
    Controller.run_all(engine=engine)

We import create_engine from SQLA and Controller from DbDeclare. We create an engine, and make sure to provide a user with admin privileges (you might need to adjust your cluster url relative to the example). Pass the engine to the run_all method, and we're done! It'll create all the entities and grant all privileges.

The entire, complete file:

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

from dbdeclare.controller import Controller
from dbdeclare.data_structures import GrantOn, Privilege
from dbdeclare.entities import Database, DatabaseContent, Role, Schema


class ExampleBase(DeclarativeBase):
    pass


class Article(ExampleBase):
    __tablename__ = "article"
    id: Mapped[int] = mapped_column(primary_key=True)


class Keyword(ExampleBase):
    __tablename__ = "keyword"
    id: Mapped[int] = mapped_column(primary_key=True)


class Cluster(ExampleBase):
    __tablename__ = "cluster"
    id: Mapped[int] = mapped_column(primary_key=True)


class BadRequest(ExampleBase):
    __tablename__ = "bad_request"
    __table_args__ = {"schema": "log"}
    id: Mapped[int] = mapped_column(primary_key=True)


class GoodRequest(ExampleBase):
    __tablename__ = "good_request"
    __table_args__ = {"schema": "log"}
    id: Mapped[int] = mapped_column(primary_key=True)


def declare_stage(stage: str) -> None:
    db = Database(name=stage)

    # "groups" aka non-login roles
    etl_writer = Role(name=f"{stage}_etl_writer")
    ml_writer = Role(name=f"{stage}_ml_writer")
    reader = Role(name=f"{stage}_reader")

    # "users" aka login roles
    Role(name=f"{stage}_etl", login=True, password="fake", in_role=[etl_writer, reader])
    Role(name=f"{stage}_ml", login=True, password="fake", in_role=[ml_writer, reader])

    # create extra schemas
    log_schema = Schema(name="log", database=db)

    # create db content
    content = DatabaseContent(name="main", database=db, sqlalchemy_base=ExampleBase, schemas=[log_schema])

    # grant privileges
    etl_writer.grant(
        grants=[
            GrantOn(
                privileges=[Privilege.INSERT, Privilege.UPDATE],
                on=[content.tables[Article.__tablename__], content.tables[Keyword.__tablename__]],
            )
        ]
    )
    ml_writer.grant(
        grants=[GrantOn(privileges=[Privilege.INSERT, Privilege.UPDATE], on=[content.tables[Cluster.__tablename__]])]
    )
    reader.grant(
        grants=[
            GrantOn(
                privileges=[Privilege.SELECT],
                on=[
                    content.tables[Article.__tablename__],
                    content.tables[Keyword.__tablename__],
                    content.tables[Cluster.__tablename__],
                ],
            )
        ]
    )

    # create log role and grant privileges if not test stage
    if stage != "test":
        log_role = Role(
            name=f"{stage}_logger",
            grants=[
                GrantOn(privileges=[Privilege.USAGE], on=[log_schema]),
                GrantOn(
                    privileges=[Privilege.INSERT, Privilege.SELECT, Privilege.UPDATE],
                    on=[content.tables[BadRequest.__tablename__], content.tables[GoodRequest.__tablename__]],
                ),
            ],
        )
        Role(name=f"{stage}_api", login=True, password="fake", in_role=[log_role, reader])


def main() -> None:
    # declare stages
    stages = ["test", "dev", "prod"]
    for stage in stages:
        declare_stage(stage)

    # create engine with admin user and default database
    engine = create_engine(url="postgresql+psycopg://postgres:postgres@127.0.0.1:5432/postgres")
    # create all entities and grant all privileges
    Controller.run_all(engine=engine)


if __name__ == "__main__":
    main()

If you run this (with an active Postgres instance/cluster), it should create all the databases, roles, schemas, and tables. It should also grant all the privileges declared. You can test via psql:

> psql -h 127.0.0.1 -U dev_api -d dev

password for user dev_api: ***

dev=> SELECT * FROM log.good_request;
 id 
----
(0 rows)

There's nothing in there, but the database, schema, role, and access privileges are all present! Woo!

That's all for now. The future holds more features, and this documentation will be updated as they're added.