Events

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.

Get more information about the Events API here

Create

Creates a list of events provided in the body of the call. 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
    },
    ...
]

The following code creates a simple list of events.

from genomcore.client import GenomcoreApiClient
api = GenomcoreApiClient(token="A_VALID_TOKEN", refresh_token="A_VALID_REFRESH_TOKEN")

body = [
    {
        "owner": "user",
        "collection": "test",
        "start": "2021-01-01T00:00:00Z",
        "value": "test",
        "type": "test",
        "source": "test",
        "externalId": "test"
    },
    {
        "owner": "user",
        "collection": "test",
        "start": "2021-01-01T00:00:00Z",
        "value": "test",
        "type": "test",
        "source": "test",
        "externalId": "test"
    }
    ...
]

api.events.create_events(body = body, chunksize = 10000, max_retries = 3)

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.

Query

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

The following code deletes all the events in the specified collection.

from genomcore.client import Genomcore

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

api.events.delete_events(collection = "some_collection_of_events")