cattle_grid.account.server¶

cattle_grid.account.server ¶

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)

event_history `async` ¶

event_history(
    account: CurrentAccount,
    start_from: str,
    session: SqlSession,
) -> EventHistoryResponse

Not implemented yet

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

@account_router.get("/history", operation_id="event_history")
async def event_history(
    account: CurrentAccount, start_from: str, session: SqlSession
) -> EventHistoryResponse:
    """Not implemented yet"""

    uuid = UUID(start_from)

    response = await session.scalars(
        select(EventHistory)
        .where(EventHistory.account == account)
        .where(EventHistory.id > uuid)
    )

    events = [
        EventInformation(data=x.data, event_type=x.event_type, actor=x.actor)
        for x in response
    ]

    return EventHistoryResponse(events=events)

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",
    response_class=EventStreamResponse,
)
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.""",
    )

EventHistoryResponse ¶

Bases: BaseModel

Returns the history of events

Parameters:

Name	Type	Description	Default
`events`	`list[EventInformation]`		required

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

class EventHistoryResponse(BaseModel):
    """Returns the history of events"""

    events: list[EventInformation]

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()

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

cattle_grid.account.server¶