tgpm package

Subpackages

Submodules

tgpm.exc module

exception tgpm.exc.UserNotSet

Bases: Exception

tgpm.model_permission_generator module

class tgpm.model_permission_generator.ModelPermissionGenerator(user_model: type[DeclarativeBase])

Bases: object

Class to generate permission models for resources.

PERMISSION_SUFFIX = 'permissions'
generate_permission_model_for(model: type[DeclarativeBase]) type[PermissionModelProtocol]

Generate a permission model for the specified resource model.

Parameters:

model (Type[DeclarativeBase]) – The resource model class.

Returns:

The generated permission model class.

Return type:

Type[PermissionModelProtocol]

generate_permission_models_for_models(models: list[type[DeclarativeBase]]) dict[type[DeclarativeBase], PermissionModelProtocol]
class tgpm.model_permission_generator.PermissionModelProtocol(*args, **kwargs)

Bases: Protocol

resource: RelationshipProperty
resource_id: MappedColumn
scope: Mapped[PermissionScope]
type_: Mapped[PermissionType]
user_id: MappedColumn
class tgpm.model_permission_generator.PermissionScope(value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Enumeration representing the scope of permissions.

CHILDREN_ALLOWED = 'children_allowed'
CHILDREN_DENIED = 'children_denied'
DENIED = 'denied'
SELF = 'self'
class tgpm.model_permission_generator.PermissionType(value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

READ = 'read'
WRITE = 'write'
tgpm.model_permission_generator.get_foreign_key_for(model: type[DeclarativeBase]) ForeignKey

generate a ForeignKey on the provided model primary key

Parameters:

model – the model to get the primary key from

Returns:

a foreign key on the primary key of model

tgpm.model_permission_generator.get_primary_key(model: type[DeclarativeBase]) Column

tgpm.permission_adder module

class tgpm.permission_adder.PermissionAdder(repository: AsyncRepository, permission_scope: PermissionScope = PermissionScope.SELF, permission_type: PermissionType = PermissionType.READ, user_id=None)

Bases: Generic

Class to add permissions to a repository in a human-readable way.

for_(user_id: Any) PermissionAdder[T]

Specify the user ID for whom the permission is being added.

Parameters:

user_id (Any) – The user ID.

Returns:

Self

Return type:

PermissionAdder[T]

async where(*, resource_id: Any)

Specify the resource ID and add the permission to the repository.

Parameters:

resource_id (T2) – The resource ID.

Returns:

The result of creating the permission.

Return type:

T3

Raises:

UserNotSet – If user ID is not set before calling this method.

with_(*, scope: PermissionScope = None, permission_type: PermissionType = None) PermissionAdder[T]

Specify the scope and/or permission type.

Parameters:
Returns:

Self

Return type:

PermissionAdder[T1, T2, T3]

async tgpm.permission_adder.create_permission(repository: AsyncRepository, user_id, resource_id, permission_scope: PermissionScope, permission_type: PermissionType)

Create a permission in the repository.

Parameters:
  • repository (AsyncRepository) – The AsyncRepository instance for permission model.

  • user_id – The user ID.

  • resource_id – The resource ID.

  • permission_scope (PermissionScope) – The permission scope.

  • permission_type (PermissionType) – The permission type.

Returns:

The result of creating the permission.

Return type:

T3

tgpm.permission_validator module

exception tgpm.permission_validator.PermissionDenied(resource_type: type[DeclarativeBase], resource_id, user_id, permission_type: PermissionType)

Bases: Exception

Exception raised when permission is denied.

class tgpm.permission_validator.ValidateThat(repository: ResourcePermissionRepository, user_id: Any = None)

Bases: Generic

async can_read(resource_id: Any, *, user_id: Any = None) None

Check if the user has read permission on the resource.

Parameters:
  • resource_id (T2) – The resource ID.

  • user_id (T1, optional) – The user ID, defaults to None.

Raises:

PermissionDenied – If permission is denied.

async can_write(resource_id: Any, *, user_id: Any = None) None

Check if the user has write permission on the resource.

Parameters:
  • resource_id (T2) – The resource ID.

  • user_id (T1, optional) – The user ID, defaults to None.

Raises:

PermissionDenied – If permission is denied.

tgpm.registry module

class tgpm.registry.Registry

Bases: dict[type[DeclarativeBase], PermissionModelProtocol]

tgpm.tgpm module

class tgpm.tgpm.AsyncConnectedTGPM(tgpm: TGPM, session: AsyncSession)

Bases: Generic

Class representing an instance of the permission system with a database session allowing it to interact with the permission system

add_permission_on(resource_type: type[T]) PermissionAdder

Add permission for a specific resource type.

Parameters:

resource_type (type[DeclarativeBase]) – The resource type.

Returns:

The permission adder.

Return type:

PermissionAdder

async filter(user_id: Any, resources: list[T], permission_type: PermissionType = PermissionType.READ) list[T]

filter the list of resource returning only those that the given user can access

Parameters:
  • user_id – the id of the user to check its permissions

  • resources – the list of resources to filter

  • permission_type – the type of permission to use for searching the allowed resource

Returns:

a filtered list of resource with only those that the given user can read/write (depending on permission_type)

get_permission_validator(resource_type: type[T]) ValidateThat

Get the permission validator for a specific resource type.

Parameters:

resource_type (type[DeclarativeBase]) – The resource type.

Returns:

The permission validator.

Return type:

ValidateThat

get_repository_for(resource_type: type[T]) AsyncRepository

Get the repository for a specific resource type.

Parameters:

resource_type (type[DeclarativeBase]) – The resource type.

Returns:

The async repository.

Return type:

AsyncRepository

get_resource_permission_repository(resource_type: type[T]) ResourcePermissionRepository

get a permission repository for the given resource_type

Parameters:

resource_type (type[DeclarativeBase]) – The resource type.

Returns:

The permission adder.

Return type:

ResourcePermissionRepository[T]

class tgpm.tgpm.TGPM(user_model: type[DeclarativeBase], model_generator: ModelPermissionGenerator)

Bases: object

Class representing the system for managing permissions. This class handle te definition of the underlying permission system.

generate_permission_for_all_models()

Generate permissions models for all models defined in the sqlalchemy registry of the user model. Creating a new table for each one.

Must be called after all models are defined or models created after must be manually included with get_permission_model_for.

get_permission_model_for(resource_type: type[DeclarativeBase]) PermissionModelProtocol

Get the permission model for a specific resource type. or generate one if not defined already.

Parameters:

resource_type (type[DeclarativeBase]) – The resource type.

Returns:

The permission model.

Return type:

PermissionModelProtocol

use(session: AsyncSession) AsyncConnectedTGPM

Generate an instance of AsyncConnectedDradis allowing interaction with the permission system.

Parameters:

session (AsyncSession) – The async session.

Returns:

An instance of AsyncConnectedDradis.

Return type:

AsyncConnectedTGPM

tgpm.tgpm.get_tgpm(user_model: type[DeclarativeBase]) TGPM

factory to create a tgpm instance from a user model

Parameters:

user_model (type[DeclarativeBase]) – the user model to use for generating permissions

Returns:

a tgpm system instance

Return type:

TGPM

Module contents