saigon

logutils

A set of utilities to enable structured context-aware logging in your application.

This module provides multiple utilities to incorporate context-aware logging in as many parts of your code as possible. The main construct, logcontext, allows you to create a local scoped context to define key-value items that will be appended all log messages generated under this context. Key-value items are automatically removed after exiting the context, hence log messages outside of it will not contain them.

See the following basic example:

import logging
import logutils
...

logutils.enable_log_context()
logger = logging.getLogger()
...

logger.info('before context')
with logutils.log_context():
    set_log_context(my_key='value')
    logger.info('in context')

logger.info('after context')

The above code will produce the following log messages (omitting some values):

{"message": "before context", "level": "INFO", ...}
{"message": "in context", "my_key': "value", "level": "INFO", ...}
{"message": "after context", "level": "INFO", ...}

Note how only the second log–the one in the context and after set the key–contains the key item my_key. Additionally, the format of the log messages is JSON, a feature also provided as part of enabling context logging.

Using Context Logging

Three steps are needed to enable and use context logging:

Enable structured context logging for the built-in Python logger with enable_log_context.
Create a local scoped context to add log key items.
Set/Unset key items under your context.

Understanding Scopes

Logging contexts are implemented as contextlib::AbstractContextManager using contextvars::ContextVar. This implementation allows to create self-managed contexts with top-down items visibility. That is, key items set in an outer scope are passed down to its inner scopes, which in addition can not only set newer keys but also override any of the keys from the outer scope. Then, when the inner scope exits, the outer scope is restored to the original keys that were set prior to the creation of the inner scope.

Let’s look at the following example to understand this behavior. We’ll use explicit context creation for clarity, but the behavior is equally applicable with the decorator.

For example:

with logutils.logcontext():
    logger.info('start of outer scope')
    set_log_context(outer1='value1', outer2='value2')
    with logutils.logcontext():
        # start of inner log context scope
        set_log_context(inner='value3', outer2='override')
        logger.info('showing inner scope')
    logger.info('end of outer scope')

The generated log messages (some content omitted):

{"message": "start of outer scope", "outer1": "value1", "outer2":"value2" }
{"message": "showing inner scope", outer1": "value1", "outer2": "override", "inner": "value3" }
{"message": "start of outer scope", "outer1": "value1", "outer2": "value2" }

Notice how the first and third logs have the same key items, even though the inner scope overrides one of them (outer2) as well sets a new item (inner), as shown in the second log. These changes are shown in the second log, which also contains the outer scope item

Multithreading

The use of the logging context is thread-safe since a new underlying context is created for each thread. The aspect to keep in mind is that children threads do not inherit the key items of their parent’s thread, nor new items defined by children threads are passed to the parent.

This contrasts with coroutines, in which the context is passed down from the parent task to its children, but no the other way around. This is the similar behavior shown above in regard to outer and inner scopes.

Async Support

In order to use context logging in your async functions use the asynclogcontext decorator, which provides the same construct as logcontext but for async functions.

class saigon.logutils.asynclogcontext

Bases: AbstractAsyncContextManager, AsyncContextDecorator

Same as logcontext but for use on async functions.

Example:

@asynclogcontext()
async def my_async_function():
    set_log_context(my_var='value')

See:: logcontext

saigon.logutils.enable_log_context(log_prefix=None)

Enables context logging support in your application. Typically, you’ll call this function in the entry point of your application, before generating any logs.

Enabling logging context also enables JSON formatting for the log messages. This is currently mandatory in order to use logging context.

Parameters:: log_prefix (Optional[str]) – Optional prefix to be added to all logs, shown as a prompt delimited by colon before the JSON message.

class saigon.logutils.logcontext

Bases: AbstractContextManager, ContextDecorator

Creates a local scoped context to set log key items.

There are two ways you can enable a managed scope:

Decorating your function with logcontext() (easiest and recommended):

Example:

@logcontext()
def my_function():
    set_log_context(my_var='value')

This method automatically generates a self-managed context applicable to the entire function scope. You can use set_log_contex() and unset_log_context() operations to add/remove keys, respectively, as needed.

Explicitly creating the scoped context anywhere in your code:

def my_function():
    with logutils.logcontext() as lc:
        ...
        lc.set(my_key='value')
        ...

Calling set_log_context() and unset_log_context() under the scope is equivalent as lc.set() and lc.unset().

The second method gives you more granularity defining the scopes at the expense of a bit more code overhead.

set(**kwargs)

Sets in the log context the specified key items as keywords:

lc.set(key1='value1`, key2='value2`)

The provided keys are set for the current context scope. Setting existing keys will override their value for the current context, and their previous value will be restored upon scope finalization.

Parameters:: kwargs – Custom keyword arguments representing the log keys. The value can be anything that is JSON serializable.

unset(*args)

Removes the specified comma-separated list of key values, represented as strings:

lc.unset('key1', 'key2')

Removing a key from the current scope does not remove it from its parent/outer scope if it also sets it there.

Parameters:: args – List of keys to be removed as string arguments.

saigon.logutils.set_log_context(**kwargs)

Sets a list key items into the current log context.

See:: logcontext::set

saigon.logutils.unset_log_context(*args)

Removes a set of key items from the current log context.

See:: logcontext::unset

model

class saigon.model.BaseModelNoExtra(**data)

Bases: BaseModel

BaseModel with extra fields forbidden by default.

This configuration ensures that any input data containing fields not explicitly defined in the model will raise a validation error. It also sets use_enum_values to True for convenience.

class saigon.model.BasicRestResponse(**data): Bases: BaseModel

class saigon.model.DataSet(**data)

Bases: BaseModel, Generic

A generic data set model containing a list of items of a specified ModelType.

data

A list of data items. Defaults to an empty list.

Type:: List[ModelType]

class saigon.model.EmptyContent(**data)

Bases: BaseModel

A Pydantic model designed to handle empty or null content gracefully. It transforms None input into an empty dictionary.

class saigon.model.QueryDataPaginationToken(**data)

Bases: BaseModelNoExtra

Represents a pagination token for querying data, typically used for cursor-based pagination.

query_id

An identifier for the specific query. It must allow to retrieve the originating paginated query (e.g, an encoded set of params).

Type:: str

next_token

The token indicating the starting point for the next page of results. This is typically an offset integer.

Type:: str | int

classmethod from_offset(query_id, offset)

Creates a QueryDataPaginationToken instance from a query ID and an integer offset.

Parameters:

query_id (str) – An identifier for the specific query.
offset (int) – The integer offset to use as the next token.

Returns:

A new QueryDataPaginationToken instance.

Return type:

Self

property offset: int

Converts the next_token to an integer offset.

Returns:: The next_token as an integer offset.
Return type:: int

class saigon.model.QueryDataParams(**data)

Bases: BaseModelNoExtra, Generic

Parameters for querying data, supporting max count and either a pagination token or a query selection model.

max_count

The maximum number of results to return. Defaults to None.

Type:: Optional[int]

query

Either a pagination token or a detailed query selection model. Defaults to None.

Type:: Optional[QueryDataPaginationToken | QuerySelection]

classmethod camelcase_keys(object_dict)

Recursively converts all string keys in a dictionary to camelCase.

Parameters:: object_dict (Dict[str, Any]) – The dictionary whose keys are to be converted.
Returns:: A new dictionary with keys converted to camelCase.
Return type:: Dict[str, Any]

decode_query_selection(selection_type)

Decodes the query_id from this object’s pagination token back into a QuerySelection model.

This method assumes that the query_id of the pagination token contains a base64 encoded JSON string of the original query selection.

Parameters:: selection_type (Type[QuerySelection]) – The Pydantic model type to which the decoded selection should be cast.
Returns:: The decoded QuerySelection model.
Return type:: QuerySelection

encode_query_selection()

Encodes the query_selection into a URL-safe base64 string.

This method is typically used to convert a complex query selection into a string that can be used as a query_id in a pagination token.

Returns:: The URL-safe base64 encoded string of the query selection.
Return type:: str

has_max_count()

Checks if a maximum count is specified in the query parameters.

Returns:: True if max_count is not None, False otherwise.
Return type:: bool

has_pagination_token()

Checks if the query member represents a pagination token.

Returns:: True if query is a QueryDataPaginationToken instance, False otherwise.
Return type:: bool

has_query_selection()

Checks if the query parameter represents a query selection model (and not a pagination token).

Returns:: True if query is present and not a pagination token, False otherwise.
Return type:: bool

property pagination_token: QueryDataPaginationToken | None

Returns the query as a QueryDataPaginationToken if it contains a pagination token, otherwise returns None.

Returns:: The pagination token if present, otherwise None.
Return type:: Optional[QueryDataPaginationToken]

property query_selection: QuerySelection | None

Returns the query as a QuerySelection model if it represents a query selection, otherwise returns None.

Returns:: The query selection model if present, otherwise None.
Return type:: Optional[QuerySelection]

classmethod to_camelcase(value)

Converts a snake_case string to upper camelCase.

Parameters:: value (str) – The snake_case string to convert.
Returns:: The camelCase string.
Return type:: str

url_params_dict(camel_case=True)

Generates a dictionary of URL parameters from the query data parameters. Keys can optionally be converted to CamelCase.

Parameters:: camel_case (bool) – Whether parameter names are formatted as CamelCase.
Returns:: A dictionary suitable for URL query parameters.
Return type:: Dict[str, Any]

class saigon.model.QueryDataResult(**data)

Bases: DataSet

Represents the result of a data query, including the data set and an optional pagination token for subsequent queries.

pagination_token

An optional pagination token for fetching the next page of results.

Type:: Optional[QueryDataPaginationToken]

class saigon.model.Range(**data)

Bases: BaseModelNoExtra, Generic

A generic range model with a start and an end value. It ensures that the start value is not greater than the end value.

start

The starting value of the range. Defaults to None.

Type:: RangeType

end

The ending value of the range. Defaults to None.

Type:: RangeType

property length: Delta

Returns the length of the range.

This property is expected to be used with types that support subtraction, like datetime or timedelta.

Returns:: The difference between the end and start values.
Return type:: Delta

validate()

Pydantic model validator to ensure that the start value is not greater than the end value.

Raises:: ValueError – If start is greater than end.
Returns:: The validated Range instance.
Return type:: Self

class saigon.model.TimeRange(**data)

Bases: Range[datetime]

A specific Range implementation for datetime objects, representing a time interval.

start

The start time of the range. Defaults to datetime(1, 1, 1, 0, 0, 0). Serialized as ‘start_time’.

Type:: datetime

end

The end time of the range. Defaults to the current datetime if not provided. Serialized as ‘end_time’.

Type:: datetime

length()

Returns the length of the TimeRange as a timedelta.

Overrides:: Range.length: Provides a concrete return type for timedelta.

Returns:: The duration between the end and start datetimes.
Return type:: timedelta

sftp

class saigon.sftp.SftpCredentials(**data): Bases: BaseModelNoExtra

utils

class saigon.utils.Environment(**kwargs)

Bases: ABC, BaseModel

Abstract base class for environment configurations.

This class provides functionality to load environment variables into Pydantic models and set model attributes as environment variables. It allows for flexible handling of configuration based on both passed arguments and system environment variables.

model_config

Pydantic configuration allowing extra fields.

Type:: ConfigDict

Example

Consider a configuration for a database:

class DatabaseConfig(Environment):
    HOST: str
    PORT: int = 5432
    USER: str

# If DB_HOST and DB_USER are set in environment variables
# e.g., export DB_HOST="localhost", export DB_USER="admin"
db_config = DatabaseConfig(PORT=5433)
print(db_config.HOST) # Output: 'localhost'
print(db_config.PORT) # Output: 5433
print(db_config.USER) # Output: 'admin'

# Set these values back to environment variables
db_config.setvars()
print(os.getenv('HOST')) # Output: 'localhost'

setvars()

Sets the instance’s attribute values as environment variables.

For each field defined in the model, if the instance has a value for that field, it will be converted to a string and set as an environment variable with the field’s name.

Returns:: The current instance, allowing for method chaining.
Return type:: Self

class saigon.utils.EnvironmentRepository(*args, **kwargs): Bases: KeyValueRepository

class saigon.utils.NameValueItem(name: str, value: ValueType)

Bases: NamedTuple, Generic

Represents a simple name-value pair.

name

The name of the item.

Type:: str

value

The value associated with the name.

Type:: ValueType

name: str: Alias for field number 0

value: TypeVar(ValueType): Alias for field number 1

class saigon.utils.NodeEntity(**data)

Bases: BaseModel, Generic[NodeEntityType]

Represents a node in a tree-like structure, holding an entity.

This class allows for building hierarchical data structures where each node contains a Pydantic model (entity), and can have a parent and multiple children.

entity

The Pydantic model payload for this node.

Type:: NodeEntityType

parent

The parent node in the hierarchy. Defaults to None.

Type:: Optional[Self]

children

A list of child nodes. Defaults to an empty list.

Type:: List[Self]

Example:

class Document(BaseModel):
    id: str
    name: str

root_doc = Document(id="1", name="Root Document")
child_doc_1 = Document(id="2", name="Child Document 1")
child_doc_2 = Document(id="3", name="Child Document 2")

root_node = NodeEntity(entity=root_doc)
child_node_1 = NodeEntity(entity=child_doc_1)
child_node_2 = NodeEntity(entity=child_doc_2)

root_node.add_child(child_node_1)
root_node.add_child(child_node_2)

print(child_node_1.parent.entity.name) # Output: Root Document

# Traverse and print node names
def print_node_name(node: NodeEntity):
    print(f"Node: {node.entity.name}")

root_node.traverse(print_node_name)
# Expected Output:
# Node: Root Document
# Node: Child Document 1
# Node: Child Document 2

add_child(node)

Adds a child node to the current node.

Also sets the parent attribute of the added child node to this node.

Parameters:: node (Self) – The node to add as a child.

serialize_parent(parent, _info)

Serializes the ‘parent’ field to its ‘name’ attribute if it exists.

This custom serializer is used by Pydantic when serializing a NodeEntity instance. It prevents recursive serialization of the parent object and instead just includes its name.

Parameters:

parent (Self) – The parent entity being serialized.
_info – Pydantic’s SerializationInfo object (unused here but required).

Returns:

The name of the parent entity if it has a ‘name’ attribute,: otherwise None.

Return type:

Optional[str]

traverse(visitor)

Performs a depth-first traversal of the node and its children.

Applies a visitor function to the current node and then recursively to all its children.

Parameters:: visitor (Callable[[Self], Any]) – A callable that takes a NodeEntity instance as input.

fflags

iter

saigon.iter.contains(item_list, condition=<function is_true_or_valid>)

Checks whether item_list contains at least one element that meets the specified condition.

Parameters:

item_list (Iterable[TypeVar(ItemType)]) – Iterable of items of type ItemType
condition (Optional) – Condition logic to apply to an item. Default is set
is_true_or_valid (to)

Return type:

bool

Returns:

True if at least one item meets the condition

saigon.iter.first(item_list, condition=<function is_true_or_valid>)

Returns the first item in item_list that meets the specified condition.

Parameters:

item_list (Iterable[TypeVar(ItemType)]) – Iterable of items of type ItemType
condition (Optional) – Condition logic to apply to an item. Default is set
is_true_or_valid (to)

Return type:

Optional[TypeVar(ItemType)]

Returns:

First item that meets the condition

saigon.iter.is_true_or_valid(item)

Checks the input value is not none or if it’s True

Parameters:: item (TypeVar(ItemType)) – Input value to check against
Return type:: bool
Returns:: True if the input value is not None, or its actual value is ItemType is bool

saigon.iter.select(item_list, condition=<function is_true_or_valid>)

Returns the subset of items that meets the specified condition.

Parameters:

item_list (Iterable[TypeVar(ItemType)]) – Iterable of items of type ItemType
condition (Optional) – Condition logic to apply to an item. Default is set
is_true_or_valid (to)

Return type:

Iterable[TypeVar(ItemType)]

Returns:

An iterable with the selected items