rss.storage

rss.storage

create_id(base_url)

Creates a new id for an object

Source code in rss/storage.py

def create_id(base_url: str) -> str:
    """Creates a new id for an object"""

    return urljoin(base_url, str(uuid7()))

create_object(connection, config, obj) async

Given an ActivityPub object without id, it is assigned an id, syndicated via cattle_grid, and stored.

Parameters:

Name	Type	Description	Default
broker			required
base_url		The start of the URI for the object	required
obj	dict	The base object being created	required

Returns:

Type	Description
str	The URI of the created object

Source code in rss/storage.py

async def create_object(connection, config: Config, obj: dict) -> str:
    """
    Given an ActivityPub object without id, it is assigned
    an id, syndicated via cattle_grid, and stored.

    :param broker:
    :param base_url: The start of the URI for the object
    :param obj: The base object being created
    :returns: The URI of the created object
    """
    object_id = create_id(config.object_url)

    obj["id"] = object_id
    actor_id = obj.get("attributedTo")
    activity_factory = ActivityFactory({"id": actor_id})

    activity = activity_factory.create(obj).build()
    activity["id"] = f"{object_id}#create"

    await connection.trigger(
        "send_message",
        {
            "actor": actor_id,
            "data": activity,
        },
    )

    await StoredObject.create(id=object_id, data=obj)

    return object_id

retrieve_object(object_id) async

Given an URI returns the corresponding ActivityPub object otherwise none.

Parameters:

Name	Type	Description	Default
object_id	str		required

Returns:

Type	Description
dict | None

Source code in rss/storage.py

async def retrieve_object(object_id: str) -> dict | None:
    """Given an URI returns the corresponding ActivityPub object
    otherwise none.

    :param object_id:
    :returns:
    """
    result = await StoredObject.get_or_none(id=object_id)

    if result is None:
        return None

    return result.data

with_database(db_url='sqlite://:memory:') async

Context manager that opens the database connection

Parameters:

Name	Type	Description	Default
db_url		Database connection string	'sqlite://:memory:'

Source code in rss/storage.py

@asynccontextmanager
async def with_database(db_url="sqlite://:memory:"):
    """Context manager that opens the database connection

    :param db_url: Database connection string
    """
    await Tortoise.init(
        db_url=db_url,
        modules={
            "rss_models": ["rss.models"],
        },
    )
    await Tortoise.generate_schemas()

    yield "db"

    await Tortoise.close_connections()