RoleManager

RoleManager manages a collection of Role objects, allowing creation and permission management for multiple roles in the PostgreSQL database.

Source code in pum/role_manager.py

class RoleManager:
    """
    RoleManager manages a collection of Role objects,
    allowing creation and permission management
    for multiple roles in the PostgreSQL database.
    """

    def __init__(self, roles=list[Role] | list[dict]) -> None:
        """Initialize the RoleManager class.:
        Args:
            roles: List of roles or dictionaries defining roles.
            Each role can be a dictionary with keys 'name', 'permissions', and optional 'description' and 'inherit'.
        """
        if isinstance(roles, list) and all(isinstance(role, dict) for role in roles):
            self.roles = {}
            for role in roles:
                _inherit = role.get("inherit")
                if _inherit is not None:
                    if _inherit not in self.roles:
                        raise ValueError(
                            f"Inherited role {_inherit} does not exist in the already defined roles. Pay attention to the order of the roles in the list."
                        )
                    role["inherit"] = self.roles[_inherit]
                self.roles[role["name"]] = Role(**role)
        elif isinstance(roles, list) and all(isinstance(role, Role) for role in roles):
            _roles = copy.deepcopy(roles)
            self.roles = {role.name: role for role in _roles}
        else:
            raise TypeError("Roles must be a list of dictionaries or Role instances.")

        for role in self.roles.values():
            if role.inherit is not None and role.inherit not in self.roles.values():
                raise ValueError(
                    f"Inherited role {role.inherit.name} does not exist in the defined roles."
                )

    def create_roles(
        self, connection: psycopg.Connection, grant: bool = False, commit: bool = False
    ) -> None:
        """Create roles in the database.
        Args:
            connection: The database connection to execute the SQL statements.
            grant: Whether to grant permissions to the roles. Defaults to False.
            commit: Whether to commit the transaction. Defaults to False.
        """
        for role in self.roles.values():
            role.create(connection=connection, commit=False, grant=grant)
        if commit:
            connection.commit()

    def grant_permissions(self, connection: psycopg.Connection, commit: bool = False) -> None:
        """Grant permissions to the roles in the database.
        Args:
            connection: The database connection to execute the SQL statements.
            commit: Whether to commit the transaction. Defaults to False.
        """
        for role in self.roles.values():
            for permission in role.permissions():
                permission.grant(role=role.name, connection=connection, commit=False)
        if commit:
            connection.commit()

__init__

__init__(roles=list[Role] | list[dict]) -> None

Initialize the RoleManager class.: Args: roles: List of roles or dictionaries defining roles. Each role can be a dictionary with keys 'name', 'permissions', and optional 'description' and 'inherit'.

Source code in pum/role_manager.py

def __init__(self, roles=list[Role] | list[dict]) -> None:
    """Initialize the RoleManager class.:
    Args:
        roles: List of roles or dictionaries defining roles.
        Each role can be a dictionary with keys 'name', 'permissions', and optional 'description' and 'inherit'.
    """
    if isinstance(roles, list) and all(isinstance(role, dict) for role in roles):
        self.roles = {}
        for role in roles:
            _inherit = role.get("inherit")
            if _inherit is not None:
                if _inherit not in self.roles:
                    raise ValueError(
                        f"Inherited role {_inherit} does not exist in the already defined roles. Pay attention to the order of the roles in the list."
                    )
                role["inherit"] = self.roles[_inherit]
            self.roles[role["name"]] = Role(**role)
    elif isinstance(roles, list) and all(isinstance(role, Role) for role in roles):
        _roles = copy.deepcopy(roles)
        self.roles = {role.name: role for role in _roles}
    else:
        raise TypeError("Roles must be a list of dictionaries or Role instances.")

    for role in self.roles.values():
        if role.inherit is not None and role.inherit not in self.roles.values():
            raise ValueError(
                f"Inherited role {role.inherit.name} does not exist in the defined roles."
            )

create_roles

create_roles(connection: Connection, grant: bool = False, commit: bool = False) -> None

Create roles in the database. Args: connection: The database connection to execute the SQL statements. grant: Whether to grant permissions to the roles. Defaults to False. commit: Whether to commit the transaction. Defaults to False.

Source code in pum/role_manager.py

def create_roles(
    self, connection: psycopg.Connection, grant: bool = False, commit: bool = False
) -> None:
    """Create roles in the database.
    Args:
        connection: The database connection to execute the SQL statements.
        grant: Whether to grant permissions to the roles. Defaults to False.
        commit: Whether to commit the transaction. Defaults to False.
    """
    for role in self.roles.values():
        role.create(connection=connection, commit=False, grant=grant)
    if commit:
        connection.commit()

grant_permissions

grant_permissions(connection: Connection, commit: bool = False) -> None

Grant permissions to the roles in the database. Args: connection: The database connection to execute the SQL statements. commit: Whether to commit the transaction. Defaults to False.

Source code in pum/role_manager.py

def grant_permissions(self, connection: psycopg.Connection, commit: bool = False) -> None:
    """Grant permissions to the roles in the database.
    Args:
        connection: The database connection to execute the SQL statements.
        commit: Whether to commit the transaction. Defaults to False.
    """
    for role in self.roles.values():
        for permission in role.permissions():
            permission.grant(role=role.name, connection=connection, commit=False)
    if commit:
        connection.commit()