Skip to content

cattle_grid.account.server

cattle_grid.account.server

signin async 

signin(data: SignInData) -> 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 
23
24
25
26
27
28
29
30
31
32
33
34
@router.post("/signin", operation_id="signin")
async def signin(data: SignInData) -> 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(data.name, data.password)
    if account is None:
        raise HTTPException(401)

    token = str(uuid4())
    await AuthenticationToken.create(account=account, token=token)

    return TokenResponse(token=token)

account

create_actor_method async 

create_actor_method(
    body: CreateActorRequest, account: CurrentAccount
)

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 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
@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):
    """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(body.base_url, preferred_username=body.handle)
    except DuplicateIdentifierException:
        raise HTTPException(409, "Duplicate identifier")

    await actor.fetch_related("identifiers")

    name = body.name or "from_api"

    await ActorForAccount.create(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,
) -> InformationResponse

Returns information about the server and the account.

Source code in cattle_grid/account/server/account.py 
73
74
75
76
77
78
79
80
81
82
83
@account_router.get("/info", operation_id="account_info")
async def return_account_information(
    account: CurrentAccount, method_information: MethodInformation
) -> 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(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 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
@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,
    broker: Broker,
    exchange: InternalExchange,
    transformer: Transformer,
) -> 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 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@actor_router.post("/lookup", response_model_exclude_none=True, operation_id="lookup")
async def lookup(
    body: LookupRequest,
    account: CurrentAccount,
    broker: Broker,
    exchange: InternalExchange,
    transformer: Transformer,
) -> 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(account, body.actor_id)
    if actor is None:
        raise HTTPException(400)

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

    result = await broker.publish(
        msg, routing_key="fetch_object", exchange=exchange, rpc=True
    )

    if result is None:
        return {"raw": {}}

    return await transformer({"raw": result})

trigger_action async 

trigger_action(
    method: str,
    body: PerformRequest,
    account: CurrentAccount,
    broker: Broker,
    exchange: ActivityExchange,
)

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 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
@actor_router.post("/trigger/{method}", status_code=202, operation_id="trigger")
async def trigger_action(
    method: str,
    body: PerformRequest,
    account: CurrentAccount,
    broker: Broker,
    exchange: ActivityExchange,
):
    """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(account, body.actor)

    if actor is None:
        raise HTTPException(400)

    await broker.publish(
        body.model_dump(),
        routing_key=method,
        exchange=exchange,
    )

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 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
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 
18
19
20
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 
23
24
25
26
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
required
Source code in cattle_grid/account/server/responses.py 
29
30
31
32
33
34
35
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 
4
5
6
7
8
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 
11
12
13
14
15
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")