eventit_py.logging_backends module

class eventit_py.logging_backends.BaseLoggingClient(groups: list[str], exclude_none: bool = True)

Bases: object

Base class for logging clients.

Parameters:

  • groups (list[str]) – A list of groups that the logging client belongs to.

  • exclude_none (bool, optional) – Whether to exclude None values when logging. Defaults to True.

count_events_by_query(query_dict: dict, group: str, event_type: BaseEvent) → int

Count the number of times an event has occurred based on a query dictionary for a specific group and event type.

Parameters:

  • query_dict (dict) – A dictionary where the key is the field to match and the value is the value to match.

  • group (str) – The group to search events in.

  • event_type (str) – The type of event to retrieve.

Returns:

The number of events that match the query for the specified group and event type.

Return type:

int

get_event_by_uuid(uuid_obj: str, group: str, event_type: BaseEventType) → BaseEventType

Retrieve an event by its UUID.

Parameters:

  • uuid (str) – The UUID of the event to retrieve.

  • group (str) – The group to search events in.

  • event_type (str) – The type of event to retrieve.

Returns:

The event that matches the UUID for the specified group and event type.

Return type:

BaseModel

log_message(message: BaseEvent, group: str) → None

Logs a message to the specified group.

Parameters:

  • message (BaseEvent) – The message to be logged.

  • group (str) – The group to log the message to.

Raises:

NotImplementedError – This method must be implemented in derived classes.

search_events_by_query(query_dict: dict, group: str, event_type: BaseEventType, limit: int | None = None) → List[BaseEventType]

Search events based on a query dictionary for a specific group and event type.

Parameters:

  • query_dict (dict) – A dictionary where the key is the field to match and the value is the value to match.

  • group (str) – The group to search events in.

  • event_type (BaseEventType) – The type of event to retrieve.

  • limit (int, optional) – The maximum number of events to return. Defaults to None.

Returns:

A list of events that match the query for the specified group and event type.

Return type:

List[BaseModel]

search_events_by_timestamp(start_time: datetime, end_time: datetime, group: str, event_type: BaseEventType, limit: int | None = None) → List[BaseEventType]

Search events within a specified time range for a specific group and event type.

Parameters:

  • start_time (datetime) – The start time of the search range.

  • end_time (datetime) – The end time of the search range.

  • group (str) – The group to search events in.

  • event_type (BaseEventType) – The type of event to retrieve.

  • limit (int, optional) – The maximum number of events to return. Defaults to None.

Returns:

A list of events that fall within the specified time range for the specified group and event type.

Return type:

List[BaseModel]

update_event_by_uuid(group: str, event: BaseEvent, event_type: BaseEventType) → dict[str, int]

Update an event by its UUID. Retrieves UUID from event object.

Parameters:

  • group (str) – The group to update the event in.

  • event (BaseModel) – The updated event to store.

Returns:

None

class eventit_py.logging_backends.FileLoggingClient(directory: str, groups: list[str], filename: str | None = None, exclude_none: bool = True, separate_files: bool = True)

Bases: BaseLoggingClient

Append to files from provided filepath for logging

count_events_by_query(query_dict: dict, group: str, event_type: BaseEventType) → int

Count the number of times an event has occurred based on a query dictionary for a specific group and event type.

Parameters:

  • query_dict (dict) – A dictionary where the key is the field to match and the value is the value to match.

  • group (str) – The group to search events in.

  • event_type (str) – The type of event to retrieve.

Returns:

The number of events that match the query for the specified group and event type.

Return type:

int

get_event_by_uuid(uuid_obj: UUID, group: str, event_type: BaseEventType) → BaseEventType

Retrieve an event by its UUID.

Parameters:

  • uuid_obj (uuid.UUID) – The UUID of the event to retrieve.

  • group (str) – The group to search events in.

  • event_type (str) – The type of event to retrieve.

Returns:

The event that matches the UUID for the specified group and event type.

Return type:

BaseModel

log_message(message: BaseEventType, group: str) → None

Record the message provided into a single line, on the file opened Write newline to put next message on separate line (jsonlines format) Force file to be flushed to keep consistency for now

Parameters:

message (str) – message to be logged

search_events_by_query(query_dict: dict, group: str, event_type: BaseEventType, limit: int | None = None) → List[BaseEventType]

Search events based on a query dictionary for a specific group and event type.

Parameters:

  • query_dict (dict) – A dictionary where the key is the field to match and the value is the value to match.

  • group (str) – The group to search events in.

  • event_type (str) – The type of event to retrieve.

  • limit (int, optional) – The maximum number of events to return. Defaults to None.

Returns:

A list of events that match the query for the specified group and event type.

Return type:

List[BaseEventType]

search_events_by_timestamp(start_time: datetime, end_time: datetime, group: str, event_type: BaseEventType, limit: int | None = None) → List[BaseEventType]

Search events within a specified time range for a specific group and event type.

Parameters:

  • start_time (datetime) – The start time of the search range.

  • end_time (datetime) – The end time of the search range.

  • group (str) – The group to search events in.

  • event_type (str) – The type of event to retrieve.

  • limit (int, optional) – The maximum number of events to return. Defaults to None.

Returns:

A sorted list of events that fall within the specified time range for the specified group and event type.

Return type:

List[BaseEvent]

update_event_by_uuid(group: str, event: BaseEvent, event_type: BaseEventType) → dict[str, int]

Update an event by its UUID.

Parameters:

  • uuid (str) – The UUID of the event to update.

  • group (str) – The group to update the event in.

  • event (BaseModel) – The updated event to store.

Returns:

None

class eventit_py.logging_backends.MongoDBLoggingClient(mongo_url: str, groups: list[str], exclude_none: bool = True, database_name: str | None = None)

Bases: BaseLoggingClient

Utilize MongoDB as a backend for storing log information.

This class provides a logging client that uses MongoDB as the backend for storing log information. It inherits from the BaseLoggingClient class.

mongo_url

The URL of the MongoDB server.

Type:

str

groups

A list of log groups to be used.

Type:

list[str]

exclude_none

Whether to exclude None values when logging. Defaults to True.

Type:

bool, optional

database_name

The name of the MongoDB database to use. If not provided, a default name will be used.

Type:

str, optional

count_events_by_query(query_dict: dict, group: str, event_type: BaseEventType) → int

Count the number of times an event has occurred based on a query dictionary for a specific group and event type.

Parameters:

  • query_dict (dict) – A dictionary where the key is the field to match and the value is the value to match.

  • group (str) – The group to search events in.

  • event_type (str) – The type of event to retrieve.

Returns:

The number of events that match the query for the specified group and event type.

Return type:

int

get_event_by_uuid(uuid_obj: UUID, group: str, event_type: BaseEventType) → BaseEventType

Retrieve an event by its UUID.

Parameters:

  • uuid_obj (str) – The UUID of the event to retrieve.

  • group (str) – The group to search events in.

  • event_type (str) – The type of event to retrieve.

Returns:

The event that matches the UUID for the specified group and event type.

Return type:

BaseModel

log_message(message: BaseEvent, group: str) → None

Log a message into MongoDB.

This method logs a message into MongoDB by inserting it as a single document into the configured collection.

Parameters:

  • message (BaseEvent) – The message to be logged.

  • group (str) – The log group to which the message belongs.

Raises:

ValueError – If an invalid log group is provided.

Returns:

None

reset_db()

Resets the database by dropping the current database from MongoDB.

This method drops the database specified by the _database_name attribute from MongoDB. It is important to note that this action is irreversible and will permanently delete all data in the database.

Parameters:

None –

Returns:

None

search_events_by_query(query_dict: dict, group: str, event_type: BaseEventType, limit: int | None = None) → List[BaseEventType]

Search events based on a query dictionary for a specific group and event type.

Parameters:

  • query_dict (dict) – A dictionary where the key is the field to match and the value is the value to match.

  • group (str) – The group to search events in.

  • event_type (str) – The type of event to retrieve.

  • limit (int, optional) – The maximum number of events to return. Defaults to None.

Returns:

A list of events that match the query for the specified group and event type.

Return type:

List[BaseEventType]

search_events_by_timestamp(start_time: datetime, end_time: datetime, group: str, event_type: BaseEventType, limit: int | None = None) → List[BaseEventType]

Search events within a specified time range for a specific group and event type.

Parameters:

  • start_time (datetime) – The start time of the search range.

  • end_time (datetime) – The end time of the search range.

  • group (str) – The group to search events in.

  • event_type (str) – The type of event to retrieve.

  • limit (int, optional) – The maximum number of events to return. Defaults to None.

Returns:

A sorted list of events that fall within the specified time range for the specified group and event type.

Return type:

List[BaseEvent]

update_event_by_uuid(group: str, event: BaseEvent, event_type: BaseEventType | None = None) → dict[str, int]

Update an event by its UUID.

Parameters:

  • uuid (str) – The UUID of the event to update.

  • group (str) – The group to update the event in.

  • event (BaseModel) – The updated event to store.

Returns:

None