py_abac.storage.sql package

Submodules

py_abac.storage.sql.migrations module

SQL storage migrations

class py_abac.storage.sql.migrations.Migration0To0x2x1(storage)

Bases: py_abac.storage.migration.Migration

Migration between versions 0 and 0.2.1. This migration is initial.

down()

Migrate DB schema down

up()

Migrate DB schema up

property order

Number of this migration in the row of migrations

class py_abac.storage.sql.migrations.MigrationModel(**kwargs)

Bases: sqlalchemy.orm.decl_api.Base

Migration version storage

id

version

class py_abac.storage.sql.migrations.SQLMigrationSet(storage)

Bases: py_abac.storage.migration.MigrationSet

Set of migrations for SQL Storage

last_applied()

Number of the last migration that was applied up

migrations()

Get migrations. Subclasses should defile a list of storage migrations here

save_applied_number(number)

Save the last applied up migration number

py_abac.storage.sql.model module

SQL storage policy model

class py_abac.storage.sql.model.ActionTargetModel(**kwargs)

Bases: py_abac.storage.sql.model.TargetModel, sqlalchemy.orm.decl_api.Base

Action target data model

id

target_id

uid

class py_abac.storage.sql.model.PolicyModel(**kwargs)

Bases: sqlalchemy.orm.decl_api.Base

Policy data model

classmethod from_policy(policy: py_abac.policy.policy.Policy) → py_abac.storage.sql.model.PolicyModel

Create PolicyModel from Policy object

classmethod get_filter(subject_id: str, resource_id: str, action_id: str)

Get query filter for policies matching target IDs

to_policy() → py_abac.policy.policy.Policy

Get Policy object from model instance

update(policy: py_abac.policy.policy.Policy)

Update policy model instance to match policy object

actions

json

resources

subjects

uid

class py_abac.storage.sql.model.ResourceTargetModel(**kwargs)

Bases: py_abac.storage.sql.model.TargetModel, sqlalchemy.orm.decl_api.Base

Resource target data model

id

target_id

uid

class py_abac.storage.sql.model.SubjectTargetModel(**kwargs)

Bases: py_abac.storage.sql.model.TargetModel, sqlalchemy.orm.decl_api.Base

Subject target data model

id

target_id

uid

class py_abac.storage.sql.model.TargetModel

Bases: object

Base policy target model

target_id = Column(None, String(length=248), table=None, comment='Target ID used for filtering policies')

py_abac.storage.sql.storage module

SQL Storage implementation

class py_abac.storage.sql.storage.SQLStorage(scoped_session)

Bases: py_abac.storage.base.Storage

Stores and retrieves policies from SQL database

Parameters

scoped_session – SQL Alchemy scoped session

add(policy: py_abac.policy.policy.Policy)

Store a policy

delete(uid: str)

Delete a policy

get(uid: str) → Optional[py_abac.policy.policy.Policy]

Get specific policy

get_all(limit: int, offset: int) → Generator[py_abac.policy.policy.Policy, None, None]

Retrieve all the policies within a window

get_for_target(subject_id: str, resource_id: str, action_id: str) → Generator[py_abac.policy.policy.Policy, None, None]

Get all policies for given target IDs.

update(policy: py_abac.policy.policy.Policy)

Update a policy

Module contents

SQL storage