clarity-api-sdk-python 0.3.28

A Python SDK to connect to the CTI Clarity API server.

Clarity API SDK for Python

A Python SDK for connecting to the CTI API server, with structured logging included.

Installation

pip install clarity-api-sdk-python

Logging

Logging support is built with structlog.

Set the root logger by setting the environment variable LOG_LEVEL. Otherwise, the default root logging is set to INFO.

"""Example"""

import logging

from cti.logger import initialize_logger, get_logger, ExternalLoggerConfig

initialize_logger(
    external_logger_configurations=[
        ExternalLoggerConfig(name="urllib3"),
        ExternalLoggerConfig(name="httpcore"),
        ExternalLoggerConfig(name="httpx"),
        ExternalLoggerConfig(name="httpx_auth"),
        ExternalLoggerConfig(name="httpx_retries"),
    ],
    handlers=[logging.FileHandler("app.log")]
)

logger_a = get_logger("logger_a")
logger_b = get_logger("logger_b", "WARNING")

# root_logger = logging.getLogger()
# root_logger.setLevel("DEBUG")

logger_a.info("This is info message from logger_a")
logger_a.critical("This is critical message from logger_a")

# Dynamically change the log level of logger_a to WARNING
print("\nChanging logger_a level to WARNING...\n")
logging.getLogger("logger_a").setLevel(logging.WARNING)

logger_a.info("This info message from logger_a should NOT be visible.")
logger_a.warning("This is a new warning message from logger_a.")

logger_b.info("This info message from logger_b should NOT be visible.")
logger_b.warning("This is warning message from logger_b")

# the value will be redacted
logger_b.info("password", extra={"aws_access_key_id", "AKIAIOSFODNN7EXAMPLE"})

Model

Pydantic models.

API

Singleton async client

import asyncio
from cti.api.session import initialize_async_client, close_async_client

async def main():
    await initialize_async_client()
    # ... your application logic ...
    await close_async_client()

if __name__ == "__main__":
    asyncio.run(main())

# In other modules
from cti.api.session import async_client

async def fetch_data():
    if async_client:
        response = await async_client.get(...)
        return response.json()