bovine_herd

BovineHerd(app, start_background_task_inbox=None, start_background_task_outbox=None, handle_outbox_item=handle_outbox_item, db_url='sqlite://bovine.sqlite3', authorization_adder=add_authorization)

Configures the quart app to use bovine herd. Requires a bovine_store compatible store to be available at app.config[“bovine_store”]. Configures the endpoints

• /.well-known
• /activitypub
• /endpoints
• /objects

Parameters:

Name	Type	Description	Default
app	Quart	The quart app to add the endpoints to.	required
start_background_task_inbox	Callable[[ProcessingItem, BovineStoreActor], Awaitable] | None	awaitable that asynchronously handles Activities that arrived at an inbox endpoint	None
handle_outbox_item	Callable[[ProcessingItem, BovineStoreActor], Awaitable]	awaitable that synchronously handles Activities that arrived at an outbox endpoint. This function should add the new id of the Activity to the ProcessingItem, so it can be returned in the location header.	handle_outbox_item
start_background_task_outbox	Callable[[ProcessingItem, BovineStoreActor], Awaitable] | None	awaitable that asynchronously handles Activities that arrived at an outbox endpoint	None
db_url	str	The database connection	'sqlite://bovine.sqlite3'
authorization_adder	Callable[[], Awaitable]	function that stores the performer of the request in g.requester	add_authorization

Source code in bovine_herd/bovine_herd/__init__.py

def BovineHerd(
    app: Quart,
    start_background_task_inbox: Callable[[ProcessingItem, BovineStoreActor], Awaitable]
    | None = None,
    start_background_task_outbox: Callable[
        [ProcessingItem, BovineStoreActor], Awaitable
    ]
    | None = None,
    handle_outbox_item: Callable[
        [ProcessingItem, BovineStoreActor], Awaitable
    ] = handle_outbox_item,
    db_url: str = "sqlite://bovine.sqlite3",
    authorization_adder: Callable[[], Awaitable] = add_authorization,
) -> Quart:
    """Configures the quart app to use bovine herd. Requires a bovine_store compatible
    store to be available at app.config["bovine_store"]. Configures the endpoints

    * /.well-known
    * /activitypub
    * /endpoints
    * /objects

    :param app: The quart app to add the endpoints to.
    :param start_background_task_inbox: awaitable that asynchronously handles Activities
        that arrived at an inbox endpoint
    :param handle_outbox_item: awaitable that synchronously handles Activities
        that arrived at an outbox endpoint. This function should add the new id
        of the Activity to the ProcessingItem, so it can be returned in the
        location header.
    :param start_background_task_outbox: awaitable that asynchronously handles Activities
        that arrived at an outbox endpoint
    :param db_url: The database connection
    :param authorization_adder: function that stores the performer of the request
        in g.requester
    """

    register_tortoise(
        app,
        config=tortoise_config(db_url),
        generate_schemas=True,
    )

    if not start_background_task_inbox:
        start_background_task_inbox = build_default_background_task_inbox(app)
    if not start_background_task_outbox:
        start_background_task_outbox = build_default_background_task_outbox(app)

    @app.before_serving
    async def startup():
        if "session" not in app.config:
            session = aiohttp.ClientSession()
            app.config["session"] = session
        await configure_bovine_store(app)

    @app.after_serving
    async def shutdown():
        await app.config["session"].close()

    app.register_blueprint(default_configuration)
    endpoints = build_endpoints_blueprint(
        handle_outbox_item,
        start_background_task_inbox,
        start_background_task_outbox,
    )
    app.register_blueprint(endpoints, url_prefix="/endpoints")
    app.register_blueprint(objects_blueprint, url_prefix="/objects")
    app.before_request(authorization_adder)

    return app

build_default_background_task_inbox(app)

Creates a background task runner for the inbox based on starting quart.add_background_task.

Source code in bovine_herd/bovine_herd/__init__.py

def build_default_background_task_inbox(app):
    """Creates a background task runner for the inbox based on starting
    quart.add_background_task."""

    async def start_background_task_inbox(item, actor):
        app.add_background_task(
            ProcessorList(process_inbox_item, add_to_queue).__call__, item, actor
        )

    return start_background_task_inbox

build_default_background_task_outbox(app)

Creates a background task runner for the outbox based on starting quart.add_background_task.

Source code in bovine_herd/bovine_herd/__init__.py

def build_default_background_task_outbox(app):
    """Creates a background task runner for the outbox based on starting
    quart.add_background_task."""

    async def start_background_task_outbox(item, actor):
        app.add_background_task(
            ProcessorList(process_outbox_item, add_to_queue).__call__,
            item,
            actor,
        )

    return start_background_task_outbox

add_to_queue(item, *args) async

If current.app.config[“bovine_pub_sub”] is set, adds the item to a queue

FIXME: Improve logic on how the queue is chosen FIXME: event-id logic

Source code in bovine_herd/bovine_herd/add_to_queue.py

async def add_to_queue(item: ProcessingItem, *args) -> ProcessingItem | None:
    """If current.app.config["bovine_pub_sub"] is set, adds the item to
    a queue

    FIXME: Improve logic on how the queue is chosen
    FIXME: event-id logic"""
    if "bovine_pub_sub" not in current_app.config:
        return item

    actor = args[-1]

    pubsub = current_app.config["bovine_pub_sub"]

    data_s = json.dumps(item.data)
    event = ServerSentEvent(data=data_s, event="inbox")

    if "database_id" in item.meta:
        event.id = item.meta["database_id"]

    actor = actor.actor_object.build(visibility=Visibility.OWNER)
    event_source = actor["endpoints"]["eventSource"]

    await pubsub.send(event_source, event.encode())

    return item