Group

Bases: IntPK, UpdatedAtMixin, SQLModel

Source code in fastbase/models/rbac.py

class Group(IntPK, UpdatedAtMixin, SQLModel, table=True):
    __tablename__ = 'auth_group'
    name: str = Field(max_length=20, unique=True)
    description: str = Field(max_length=199, default='')
    permissions: list[str] = Field(sa_column=Column(ARRAY(String)), default=[])

    def __repr__(self):
        return modstr(self, 'name')

    @classmethod
    async def create(cls, session: AsyncSession, *,
                     name: str,  permissions: set | None = None,
                     description: str | None = None) -> Self:
        """Create a new group. Requires group.create permission."""
        try:
            group = cls(name=name, permissions=permissions, description=description)
            session.add(group)
            await session.commit()
            await session.refresh(group)
            return group
        except IntegrityError:
            raise

    # PLACEHOLDER: To follow
    @classmethod
    async def delete(cls, name: str):
        """Delete a group. Updates cache."""
        pass

    async def add_all(self, session: AsyncSession, permissions: set[str]):
        """Append new permissions to group. Requires group.update permission."""
        self.permissions = {*self.permissions, *permissions}                # noqa
        session.add(self)
        await session.commit()

    async def reset(self, session: AsyncSession, permissions: set[str] | None = None):
        """Reset permissions. Requires group.reset permission."""
        self.permissions = permissions or []
        session.add(self)
        await session.commit()

    async def describe(self, session: AsyncSession, description: str | None = None):
        """Change group description. Requires group.update permission."""
        self.description = description or ''
        await session.commit()

    # TESTME: Untested
    @classmethod
    async def collate(cls, session: AsyncSession, nameset: set[str]) -> set[str]:
        stmt = select(cls.permissions).where(cls.name.in_(nameset))     # noqa
        edata = await session.exec(stmt)
        alldata = edata.all()

        ss = set()
        for i in alldata:
            ss.update(i)
        return ss

Attributes

__tablename__ = 'auth_group' class-attribute instance-attribute

description: str = Field(max_length=199, default='') class-attribute instance-attribute

name: str = Field(max_length=20, unique=True) class-attribute instance-attribute

permissions: list[str] = Field(sa_column=Column(ARRAY(String)), default=[]) class-attribute instance-attribute

Functions

__repr__()

Source code in fastbase/models/rbac.py

def __repr__(self):
    return modstr(self, 'name')

add_all(session, permissions) async

Append new permissions to group. Requires group.update permission.

Source code in fastbase/models/rbac.py

async def add_all(self, session: AsyncSession, permissions: set[str]):
    """Append new permissions to group. Requires group.update permission."""
    self.permissions = {*self.permissions, *permissions}                # noqa
    session.add(self)
    await session.commit()

collate(session, nameset) async classmethod

Source code in fastbase/models/rbac.py

@classmethod
async def collate(cls, session: AsyncSession, nameset: set[str]) -> set[str]:
    stmt = select(cls.permissions).where(cls.name.in_(nameset))     # noqa
    edata = await session.exec(stmt)
    alldata = edata.all()

    ss = set()
    for i in alldata:
        ss.update(i)
    return ss

create(session, *, name, permissions=None, description=None) async classmethod

Create a new group. Requires group.create permission.

Source code in fastbase/models/rbac.py

@classmethod
async def create(cls, session: AsyncSession, *,
                 name: str,  permissions: set | None = None,
                 description: str | None = None) -> Self:
    """Create a new group. Requires group.create permission."""
    try:
        group = cls(name=name, permissions=permissions, description=description)
        session.add(group)
        await session.commit()
        await session.refresh(group)
        return group
    except IntegrityError:
        raise

delete(name) async classmethod

Delete a group. Updates cache.

Source code in fastbase/models/rbac.py

@classmethod
async def delete(cls, name: str):
    """Delete a group. Updates cache."""
    pass

describe(session, description=None) async

Change group description. Requires group.update permission.

Source code in fastbase/models/rbac.py

async def describe(self, session: AsyncSession, description: str | None = None):
    """Change group description. Requires group.update permission."""
    self.description = description or ''
    await session.commit()

reset(session, permissions=None) async

Reset permissions. Requires group.reset permission.

Source code in fastbase/models/rbac.py

async def reset(self, session: AsyncSession, permissions: set[str] | None = None):
    """Reset permissions. Requires group.reset permission."""
    self.permissions = permissions or []
    session.add(self)
    await session.commit()