cattle_grid.account.server

cattle_grid.account.server

signin async

signin(
    data: SignInData, session: CommittingSession
) -> TokenResponse

Allows one to sign in to an account on cattle_grid. The response a token to be included using bearer authentication.

Source code in cattle_grid/account/server/__init__.py

@router.post("/signin", operation_id="signin")
async def signin(data: SignInData, session: CommittingSession) -> TokenResponse:
    """Allows one to sign in to an account on cattle_grid.
    The response a token to be included using bearer authentication."""
    account = await account_with_name_password(session, data.name, data.password)
    if account is None:
        raise HTTPException(401)

    token = str(uuid4())
    session.add(AuthenticationToken(account=account, token=token))

    return TokenResponse(token=token)

account

create_actor_method async

create_actor_method(
    body: CreateActorRequest,
    account: CurrentAccount,
    session: CommittingSession,
)

Allows one to create a new actor. The allowed values for base_url can be retrieved using the info endpoint.

Source code in cattle_grid/account/server/account.py

@account_router.post(
    "/create",
    status_code=201,
    operation_id="create_actor",
    responses={409: {"description": "Duplicate identifier"}},
)
async def create_actor_method(
    body: CreateActorRequest, account: CurrentAccount, session: CommittingSession
):
    """Allows one to create a new actor. The allowed values for base_url
    can be retrieved using the info endpoint."""
    try:
        actor = await create_actor(
            session, body.base_url, preferred_username=body.handle
        )
    except DuplicateIdentifierException:
        raise HTTPException(409, "Duplicate identifier")

    name = body.name or "from_api"

    session.add(ActorForAccount(account=account, actor=actor.actor_id, name=name))

    return actor_to_object(actor)

return_account_information async

return_account_information(
    account: CurrentAccount,
    method_information: MethodInformation,
    session: SqlSession,
) -> InformationResponse

Returns information about the server and the account.

Source code in cattle_grid/account/server/account.py

@account_router.get("/info", operation_id="account_info")
async def return_account_information(
    account: CurrentAccount, method_information: MethodInformation, session: SqlSession
) -> InformationResponse:
    """Returns information about the server and the account."""

    if not isinstance(method_information, list):
        logger.warning("Method information is not a list")
        method_information = []

    return await create_information_response(session, account, method_information)

stream async

stream(
    event_type: EventType,
    account: CurrentAccount,
    request: Request,
    stream_messages=Depends(get_message_streamer),
)

EventSource corresponding to all messages received by the account.

This method returns an EventSource providing server sent events.

Source code in cattle_grid/account/server/account.py

@account_router.get(
    "/stream/{event_type}",
    response_description="EventSource",
    operation_id="stream",
)
async def stream(
    event_type: EventType,
    account: CurrentAccount,
    request: Request,
    stream_messages=Depends(get_message_streamer),
):
    """EventSource corresponding to all messages received
    by the account.

    This method returns an
    [EventSource](https://developer.mozilla.org/en-US/docs/Web/API/EventSource)
    providing server sent events."""
    queue, task = stream_messages(account.name, event_type)

    return ServerSentEventFromQueueAndTask(request, queue, task)

actor

lookup async

lookup(
    body: LookupRequest,
    account: CurrentAccount,
    requester: ActivityExchangeRequester,
    session: SqlSession,
) -> dict

Looks up the resource given by uri as the actor with actor id actor_id. Here looking up the actor means that the request is signed using a private key belonging to that actor.

Source code in cattle_grid/account/server/actor.py

@actor_router.post("/lookup", response_model_exclude_none=True, operation_id="lookup")
async def lookup(
    body: LookupRequest,
    account: CurrentAccount,
    requester: ActivityExchangeRequester,
    session: SqlSession,
) -> dict:
    """Looks up the resource given by `uri` as the actor with
    actor id `actor_id`. Here looking up the actor means that
    the request is signed using a private key belonging to that actor."""

    actor = await actor_from_account(session, account, body.actor_id)
    if actor is None:
        raise HTTPException(400)

    msg = FetchMessage(actor=actor.actor, uri=body.uri)

    result = await requester(msg, routing_key="fetch")  # type: ignore
    if not isinstance(result, dict):
        return {"raw": {}}
    return result

trigger_action async

trigger_action(
    method: str,
    body: PerformRequest,
    account: CurrentAccount,
    publisher: ActivityExchangePublisher,
    session: SqlSession,
)

This method allows one to trigger asynchronous activities through a synchronous request. The basic result is that the data is posted to the ActivityExchange with the routing_key specified.

Source code in cattle_grid/account/server/actor.py

@actor_router.post("/trigger/{method}", status_code=202, operation_id="trigger")
async def trigger_action(
    method: str,
    body: PerformRequest,
    account: CurrentAccount,
    publisher: ActivityExchangePublisher,
    session: SqlSession,
):
    """This method allows one to trigger asynchronous activities
    through a synchronous request. The basic result is that
    the data is posted to the ActivityExchange with the
    routing_key specified.

    """

    actor = await actor_from_account(session, account, body.actor)

    if actor is None:
        raise HTTPException(400)

    await publisher(
        body.model_dump(),
        routing_key=method,
    )

app

Helper class to create documentation for the API

dependencies

CurrentAccount module-attribute

CurrentAccount = Annotated[
    Account, Depends(get_current_account)
]

Annotation for the current account

responses

CreateActorRequest

Bases: BaseModel

Used to create an actor for the account

Parameters:

Name	Type	Description	Default
base_url	str	Base url of the actor. The actor URI will be of the form {baseUrl}/actor/some_secret	required
handle	str | None	If present, an acct-uri of the form acct:{handle}@{domain} where domain is determined from baseUrl is created	None
name	str | None	Internal name of the actor. Used to simplify display of the actor.	None

Source code in cattle_grid/account/server/responses.py

class CreateActorRequest(BaseModel):
    """Used to create an actor for the account"""

    base_url: str = Field(
        alias="baseUrl",
        examples=["http://domain.example"],
        description="""Base url of the actor. The actor URI will be
    of the form `{baseUrl}/actor/some_secret`
    """,
    )

    handle: str | None = Field(
        None,
        examples=["alice"],
        description="""If present, an acct-uri of the form `acct:{handle}@{domain}` where domain is determined from `baseUrl` is created""",
    )

    name: str | None = Field(
        None,
        examples=["Alice"],
        description="""Internal name of the actor. Used to simplify display of the actor.""",
    )

LookupRequest

Bases: BaseModel

Parameters:

Name	Type	Description	Default
actor_id	str		required
uri	str		required

Source code in cattle_grid/account/server/responses.py

class LookupRequest(BaseModel):
    actor_id: str = Field(alias="actorId")
    uri: str

LookupResponse

Bases: BaseModel

Parameters:

Name	Type	Description	Default
raw	dict		required

Source code in cattle_grid/account/server/responses.py

class LookupResponse(BaseModel):
    model_config = ConfigDict(extra="allow")

    raw: dict

PerformRequest

Bases: BaseModel

Request send to enqueue an action

Parameters:

Name	Type	Description	Default
actor	str	The actor id, must be long to the account	required

Source code in cattle_grid/account/server/responses.py

class PerformRequest(BaseModel):
    """Request send to enqueue an action"""

    model_config = ConfigDict(extra="allow")

    actor: str = Field(examples=["http://actor.example/someId"])
    """The actor id, must be long to the account"""

actor class-attribute instance-attribute

actor: str = Field(examples=["http://actor.example/someId"])

The actor id, must be long to the account

SignInData

Bases: BaseModel

Used to sign into an account

Parameters:

Name	Type	Description	Default
name	str	Name of the account	required
password	str	Password	required

Source code in cattle_grid/account/server/responses.py

class SignInData(BaseModel):
    """Used to sign into an account"""

    name: str = Field(description="Name of the account")
    password: str = Field(description="Password")

TokenResponse

Bases: BaseModel

Returns the token to be used with Bearer authentication, i.e. add the Header Authorization: Bearer {token} to the request

Parameters:

Name	Type	Description	Default
token	str	The token	required

Source code in cattle_grid/account/server/responses.py

class TokenResponse(BaseModel):
    """Returns the token to be used with Bearer authentication, i.e.
    add the Header `Authorization: Bearer {token}` to the request"""

    token: str = Field(description="The token")

testing

account_for_test async

account_for_test(sql_session) -> Account

Fixture to create an account

Source code in cattle_grid/testing/fixtures.py

@pytest.fixture
async def account_for_test(sql_session) -> Account:
    """Fixture to create an account"""
    result = await create_account(sql_session, "alice", "alice", permissions=["admin"])
    assert result
    return result

actor_for_test async

actor_for_test(sql_session) -> Actor

Fixture to create an actor

Source code in cattle_grid/testing/fixtures.py

@pytest.fixture
async def actor_for_test(sql_session) -> Actor:
    """Fixture to create an actor"""
    actor = await create_actor(sql_session, "http://localhost/ap")

    return actor

actor_with_account async

actor_with_account(sql_session, account_for_test) -> Actor

Fixture to create an actor with an account

Source code in cattle_grid/testing/fixtures.py

@pytest.fixture
async def actor_with_account(sql_session, account_for_test) -> Actor:
    """Fixture to create an actor with an account"""
    actor = await create_actor(
        sql_session, "http://localhost/ap", preferred_username="test_actor"
    )
    await add_actor_to_account(
        sql_session, account_for_test, actor, name="test_fixture"
    )

    await sql_session.refresh(actor)

    return actor

loaded_config

loaded_config()

Ensures the configuration variables are loaded

Source code in cattle_grid/testing/fixtures.py

@pytest.fixture(autouse=True, scope="session")
def loaded_config():
    """Ensures the configuration variables are loaded"""
    global_container.load_config()
    load(global_container.config)  # type: ignore

sql_engine_for_tests async

sql_engine_for_tests()

Provides the sql engine (as in memory sqlite) for tests

This fixture has autouse=True, meaning that by importing

from cattle_grid.testing.fixtures import sql_engine_for_tests

it will run automatically. The engine is initialized in the place cattle_grid expects it.

Source code in cattle_grid/testing/fixtures.py

@pytest.fixture(autouse=True)
async def sql_engine_for_tests():
    """Provides the sql engine (as in memory sqlite) for tests

    This fixture has autouse=True, meaning that by importing

    ```python
    from cattle_grid.testing.fixtures import sql_engine_for_tests
    ```

    it will run automatically. The engine is initialized in the
    place cattle_grid expects it.
    """
    async with alchemy_database("sqlite+aiosqlite:///:memory:", echo=False) as engine:
        async with engine.begin() as conn:
            await conn.run_sync(APBase.metadata.create_all)

        yield engine

sql_session async

sql_session(session_maker_for_tests)

Returns an AsyncSession to be used by tests

Source code in cattle_grid/testing/fixtures.py

@pytest.fixture()
async def sql_session(session_maker_for_tests):
    """Returns an [AsyncSession][sqlalchemy.ext.asyncio.AsyncSession] to be used by tests"""
    async with session_maker_for_tests() as session:
        yield session