Grants

First, we'll go over the details of how grants are implemented. Then, we'll continue to build our example. You can also skip ahead to the example.

Grantable entities

The Grantable class is an internal class used by DbDeclare (aka you won't use it unless you are developing and extending the package). It defines behavior for any entity that can have privileges granted to it. This includes entities like databases, schemas, and tables. Grantable entities work in tandem with Roles to declare access privileges, like connecting to a database or selecting from a table. Ultimately, you are able to declare relationships that result in a GRANT {privileges} ON {entity_type} {entity} TO {role}.

Privileges

Privileges are a set list of access control definitions, check out the Postgres documentation for a complete overview. DbDeclare defines them all in a Python enum:

from dbdeclare.data_structures import Privilege

Use this to declare what privileges you want to when you declare a grant.

Multiple ways to grant

Grantable entities and Roles can both declare grants.

From a role

The Role class has a grant method that accepts a Sequence[GrantOn], where GrantOn can be found in data_structures:

from dbdeclare.data_structures import GrantOn

GrantOn is a dataclass that has two attributes: privileges, which is a Sequence[Privilege], and on, which is a sequence of Grantable entities to grant those privileges on.

You can also pass in a Sequence[GrantOn] when you declare a Role via the __init__ method, if it is more convenient to do so.

From a grantable entity

Anything that inherits from Grantable also has a grant method, but this one accepts a Sequence[GrantTo]. GrantTo can also be found in data_structures:

from dbdeclare.data_structures import GrantTo

GrantTo is a dataclass that has two attributes: privileges, which is a Sequence[Privilege], and to, which is a sequence of Roles to grant those privileges to.

You can also pass in a Sequence[GrantTo] when you declare any grantable entity via the __init__ method, if it is more convenient to do so.

How grants are stored

So what actually happens when you run the grant methods described above? We store them in the Role that access is granted to for easy synchronization. DbDeclare also makes sure that the order of creates and grants is correct so that execution doesn't fail. Each Role has an attribute named grants of type GrantStore, which is an alias for dict[Grantable, set[Privilege]]. This structure is easy to translate to and from Postgres, and makes sure there is a single source of truth within the code.

As always, I encourage you to peek at the source code and read the docstrings for details!

Example

Let's add grants to our example. We have declared all the entities we want to create, so now we can run some grant statements.

from dbdeclare.data_structures import GrantOn, Privilege

For our example, I've opted to grant privileges from the point of view of Roles only, which means I import GrantOn but not GrantTo.

# omitted code above

    # 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__],
                ],
            )
        ]
    )

# omitted code below

Here, we grab our references for etl_writer, ml_writer, and reader, and declare grants for each of them. We allow etl_writer to insert and update on the article and keyword table, we allow ml_writer to insert and update on the cluster table, and we allow reader to select from all three of those tables. Note that we don't have to grant usage of the default schema as (typically) usage is granted to all users by default. Depending on your set up, you may need to grant connect on the roles to each database.

# omitted code above

    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])



# omitted code below

We also need to grant insert, select, and update privileges to log_role on both tables in the log schema. Since this is a non-default schema, we first grant usage on the schema, then we grant the desired privileges on both tables. Note that this is done in the creation of the role, in contrast to the previous examples that call grant on the roles after they were declared. Do whatever makes more sense for you.

Here is the entire file now:

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

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:
    stages = ["test", "dev", "prod"]
    for stage in stages:
        declare_stage(stage)


if __name__ == "__main__":
    main()

We have declared everything we set out to declare. Nice! All that's left is to actually create all of this in the database. Let's run it!