Events Controller¶

Module containing the Events Controller class

class genomcore.controllers.events.EventsController(*args, **kwargs)¶

Bases: BaseController

Events Controller class to interact with the Events API. Mirror of the time-series controller but for events, these being basic text messages with a timestamp and some metadata.

Events API docs: https://events-api.genomcore.net/api/docs

create_events(body: List[dict], chunksize: int | None = 100000, retries: int | None = 0, chunk_details: bool | None = False, measure_time: bool | None = False) → List[dict]¶

Create events.

Parameters:

body (List[dict]) – List of python dictionaries representing the events.
chunksize (int) – Size of each chunk for batch processing. Defaults to 100000 as the max for a single chunk.
max_retries (int) – Maximum number of retries for failed requests. Defaults to 3.
chunk_details (bool) – If True, prints the first and last event of each chunk. Defaults to False.
timer (bool) – If True, prints the time taken for each chunk upload. Defaults to False.

Warning

The method only checks if the batch value is None or an empty string. It is the user responsibility to ensure that the batch value is unique for each group of events.

Note

The body should have the following list of dictionaries structure:

[
    {
        "owner": str,
        "collection": str,
        "start": str (ISO 8601 datetime),
        "end": str (ISO 8601 datetime), # Optional, must be equal or greater than start
        "value": str,
        "type": str,
        "source": str,
        "externalId": str
    },
    ...
]

Returns:: A JSON response from requests.

Example:

from genomcore.client import Genomcore

api = Genomcore(token="A_VALID_TOKEN", refresh_token="A_VALID_REFRESH_TOKEN")
api.events.create_events(body = body, chunksize = 10000, max_retries = 3)

query_events(page: int | None = 0, pageSize: int | None = 10000, body: dict = None, measure_time: bool | None = False)¶

Get events by providing MongoDB queries.

Parameters:

page (int, Optional) – Page number to query. Defaults to first page (page 0).
pageSize (int, Optional) – Number of entries per page. Defaults to 10.000.
body (dict) – Payload of the request containing MongoDB queries.

Returns:

List[dict] object from the query.

Example:

from genomcore.client import Genomcore

api = Genomcore(token="A_VALID_TOKEN", refresh_token="A_VALID_REFRESH_TOKEN")

# A valid body query in MongoDB format
# This example queries the events of owner "1" in the collection "TestCollection"
# between the dates 2024-04-01 and 2024-04-08 and sorts them by timestamp in descending order.
body = {
    "filter": {
        "owner": {"$in": [1]},
        "collection": {"$in": ["TestCollection"]}
    },
    "sort": {
        "start": -1
    },
    "beforeDate": "2024-04-01",
    "afterDate": "2024-04-08"
}
api.events.query_events(page = 2, pageSize = 500, body = body)

Delete events.

Parameters:

ids (List[str]) – List of IDs of the events to delete.
owner (str) – Owner of the events to delete.
collection (str) – Collection of the events to delete.
from_date (str) – Date from which to delete the events.
to_date (str) – Date until which to delete the events.
timer (bool) – If True, prints the time taken for delete the specified datapoints. Defaults to False.

Returns:

A JSON response with the ‘deletedCount’ number of the deleted datapoints.

Warning

If the connection is closed before receiving the response, it could be due to a large deletion that takes longer than usual (more than 60 seconds). The server might close the connection but still process the request successfully. Please verify if the data points were deleted correctly.

Example:

from genomcore.client import Genomcore

api = Genomcore(token="A_VALID_TOKEN", refresh_token="A_VALID_REFRESH_TOKEN")

# Delete events of a specific collection
api.events.delete_events(collection = "some_collection_of_events")

__firstlineno__ = 14¶

__static_attributes__ = ('_requester',)¶