cattle_grid.manage

cattle_grid.manage

AccountManager dataclass

Access for managing accounts from outside cattle_grid, e.g. by an extension

Parameters:

Name	Type	Description	Default
account	Account		required
session	AsyncSession		required

Source code in cattle_grid/manage/account.py

@dataclass
class AccountManager:
    """Access for managing accounts from outside cattle_grid, e.g.
    by an extension"""

    account: Account
    session: AsyncSession

    @staticmethod
    async def for_name_and_password(
        session: AsyncSession, name: str, password: str
    ) -> "AccountManager":
        """Returns an AccountManager for the given name and password"""
        account = await account_with_name_password(session, name, password)

        if account is None:
            raise ValueError("Account not found")

        return AccountManager(account=account, session=session)

    @staticmethod
    async def for_name(session: AsyncSession, name: str) -> "AccountManager":
        """Returns an AccountManager for a given name"""

        account = await session.scalar(select(Account).where(Account.name == name))

        if account is None:
            raise ValueError("Account not found")

        return AccountManager(account=account, session=session)

    def account_information(self) -> list[ActorInformation]:
        """Returns the actors belonging to the account"""

        return [
            actor_to_information(x)
            for x in self.account.actors
            if x.status == ActorStatus.active
        ]

    async def allowed_base_urls(self) -> list[str]:
        """Returns the list of base urls allowed for the account"""
        return await allowed_base_urls(self.session, self.account)

account_information

account_information() -> list[ActorInformation]

Returns the actors belonging to the account

Source code in cattle_grid/manage/account.py

def account_information(self) -> list[ActorInformation]:
    """Returns the actors belonging to the account"""

    return [
        actor_to_information(x)
        for x in self.account.actors
        if x.status == ActorStatus.active
    ]

allowed_base_urls async

allowed_base_urls() -> list[str]

Returns the list of base urls allowed for the account

Source code in cattle_grid/manage/account.py

async def allowed_base_urls(self) -> list[str]:
    """Returns the list of base urls allowed for the account"""
    return await allowed_base_urls(self.session, self.account)

for_name async staticmethod

for_name(
    session: AsyncSession, name: str
) -> AccountManager

Returns an AccountManager for a given name

Source code in cattle_grid/manage/account.py

@staticmethod
async def for_name(session: AsyncSession, name: str) -> "AccountManager":
    """Returns an AccountManager for a given name"""

    account = await session.scalar(select(Account).where(Account.name == name))

    if account is None:
        raise ValueError("Account not found")

    return AccountManager(account=account, session=session)

for_name_and_password async staticmethod

for_name_and_password(
    session: AsyncSession, name: str, password: str
) -> AccountManager

Returns an AccountManager for the given name and password

Source code in cattle_grid/manage/account.py

@staticmethod
async def for_name_and_password(
    session: AsyncSession, name: str, password: str
) -> "AccountManager":
    """Returns an AccountManager for the given name and password"""
    account = await account_with_name_password(session, name, password)

    if account is None:
        raise ValueError("Account not found")

    return AccountManager(account=account, session=session)

ActorManager dataclass

Access for managing actors from outside cattle_grid, e.g. by an extension

Parameters:

Name	Type	Description	Default
session	AsyncSession		required
actor_id	str	The URI of the actor to manage	required

Source code in cattle_grid/manage/actor.py

@dataclass
class ActorManager:
    """Access for managing actors from outside cattle_grid, e.g.
    by an extension"""

    session: AsyncSession
    actor_id: str = field(metadata={"description": "The URI of the actor to manage"})

    _actor_for_account: ActorForAccount | None = field(default=None)
    _actor: Actor | None = field(default=None)

    async def actor_for_account(self) -> ActorForAccount:
        if not self._actor_for_account:
            self._actor_for_account = await actor_for_actor_id(
                self.session, self.actor_id
            )

        if not self._actor_for_account:
            raise ValueError("Actor not found")

        return self._actor_for_account

    async def actor(self, session: AsyncSession) -> Actor:
        if not self._actor:
            self._actor = await session.scalar(
                select(Actor).where(Actor.actor_id == self.actor_id)
            )
            await session.refresh(self._actor, attribute_names=["identifiers"])

        if not self._actor:
            raise ValueError("Actor not found")
        return self._actor

    async def add_to_group(self, group: str):
        """Adds the actor to a group"""
        actor = await self.actor_for_account()
        await add_actor_to_group(self.session, actor, group)

    async def groups(self) -> list[str]:
        """Returns the list of groups the actor belongs to"""
        actor = await self.actor_for_account()
        return await group_names_for_actor(self.session, actor)

    async def profile(self, session: AsyncSession) -> dict:
        """Returns the actor profile"""
        return actor_to_object(await self.actor(session))

add_to_group async

add_to_group(group: str)

Adds the actor to a group

Source code in cattle_grid/manage/actor.py

async def add_to_group(self, group: str):
    """Adds the actor to a group"""
    actor = await self.actor_for_account()
    await add_actor_to_group(self.session, actor, group)

groups async

groups() -> list[str]

Returns the list of groups the actor belongs to

Source code in cattle_grid/manage/actor.py

async def groups(self) -> list[str]:
    """Returns the list of groups the actor belongs to"""
    actor = await self.actor_for_account()
    return await group_names_for_actor(self.session, actor)

profile async

profile(session: AsyncSession) -> dict

Returns the actor profile

Source code in cattle_grid/manage/actor.py

async def profile(self, session: AsyncSession) -> dict:
    """Returns the actor profile"""
    return actor_to_object(await self.actor(session))