Corekit Python Library

Corekit-py is a comprehensive utility library ported from corekit-js, designed to facilitate building Python applications with standardized integrations for Azure services, databases, logging, and other common utilities.

Table of Contents

• Features
• Installation
• Library Structure
• Dependency Mapping
• Usage Guide
  • App Configuration
  • Azure Blob Storage
  • Azure Event Hubs
  • Cosmos DB
  • Logger
  • Middleware
  • PostgreSQL
  • Redis Cache
  • Utils
• Configuration
• Contributing
• License

Features

• App Configuration: Integration with Azure App Configuration and Key Vault for centralized settings.
• Azure Blob Storage: Asynchronous client for efficient blob operations.
• Azure Event Hubs: Robust async producers and consumers with automatic checkpointing using Azure Blob Storage.
• ClickHouse: Client for high-performance querying and data streaming.
• Cloudinary: Service for image and video upload, transformation, and management.
• Cosmos DB: Repository pattern implementation with caching, soft-deletes, and async wrapper around the synchronous SDK.
• HTTP: httpx-based wrapper for resilient HTTP requests with default timeouts and pooling.
• Logger: Structured logging supporting console, Slack notifications, and OpenTelemetry (OTLP) export.
• Middleware: ASGI middleware for Firebase Authentication, Rate Limiting, and Request Sanitization (compatible with Starlette/FastAPI).
• PostgreSQL: Asynchronous client and repository pattern using asyncpg.
• Queue: Azure Storage Queue management including poison message handling.
• Redis Cache: Redis client and caching utilities supporting context-based keys and JSON serialization.
• Utils: Extensive collection of helpers for date/time, geospatial calculations, encryption, retry logic, distributed locking, and more.

Installation

pip install corekit-py

Library Structure

The library is organized into modular services within src/corekit:

src/corekit/
├── app_config/        # Azure App Configuration & Key Vault
├── azure_blob/        # Azure Blob Storage Client
├── azure_event_hubs/  # Event Hubs Producer & Consumer
├── clickhouse/        # ClickHouse Client
├── cloudinary/        # Cloudinary Service
├── cosmos/            # Cosmos DB Repository & Client
├── http/              # HTTP Client Wrapper
├── logger/            # Structured & OTLP Logger
├── middleware/        # ASGI Middleware (Auth, RateLimit)
├── postgres/          # PostgreSQL Async Client
├── queue/             # Azure Queue Storage
├── redis_cache/       # Redis Client & Cache
└── utils/             # Shared Utilities (Date, Geo, etc.)

Dependency Mapping

Corekit-py relies on several external packages. Here is how they map to the internal modules:

Corekit Module	Primary Dependencies	Description
app_config	azure-appconfiguration, azure-keyvault-secrets, azure-identity	Fetches config and secrets.
azure_blob	azure-storage-blob	Blob storage operations.
azure_event_hubs	azure-eventhub, azure-eventhub-checkpointstoreblob-aio	Event ingestion and processing.
cosmos	azure-cosmos, cachetools	Database operations and in-memory caching.
clickhouse	clickhouse-connect	ClickHouse database interaction.
cloudinary	cloudinary, httpx	Media management.
http	httpx	Async HTTP requests.
logger	opentelemetry-api, opentelemetry-sdk, opentelemetry-exporter-otlp-proto-http, httpx	Telemetry and Slack alerts.
middleware	starlette, firebase-admin	Web framework integration and Auth.
postgres	asyncpg	Async PostgreSQL driver.
queue	azure-storage-queue	Queue messaging.
redis_cache	redis	Async Redis client.
utils	cryptography, cachetools	Encryption and locking mechanisms.
Global	pydantic, pydantic-settings	Data validation and settings management.

Usage Guide

App Configuration

Load configuration from Azure App Config and Key Vault on startup.

from corekit.app_config import AppConfigClient

client = AppConfigClient(connection_string="Endpoint=...")
await client.load_configurations()

# Access a configuration value
feature_flag = client.get("my-feature-flag")
secret_value = client.get("my-secret")

Azure Blob

Upload and manage blobs asynchronously.

from corekit.azure_blob import AzureBlobClient

client = AzureBlobClient(connection_string="...")
await client.upload_blob(
    container_name="my-container",
    blob_name="file.txt",
    data="Hello World"
)

Azure Event Hubs

Consume events with checkpointing.

from corekit.azure_event_hubs import AzureEventsHubsConsumer

async def process_events(events, partition_context):
    for event in events:
        print(f"Received: {event.body_as_str()}")
        await partition_context.update_checkpoint(event)

consumer = AzureEventsHubsConsumer(
    event_hub_namespace_connection_string="...",
    event_hub_name="my-hub",
    consumer_group="$Default",
    storage_connection_string="...", # For checkpointing
    block_container_name="checkpoints"
)

await consumer.subscribe(process_events)

Cosmos DB

Use the repository pattern for standardized database access.

from corekit.cosmos import CosmosClientService, CosmosRepo

# Setup
service = CosmosClientService(endpoint="...", key="...")
client = service.get_cosmos_client()
container = client.get_database_client("my-db").get_container_client("users")

# Repository
repo = CosmosRepo(container, partition_key="userId")

# Operations
user = await repo.get_item("user-123")
await repo.upsert_item({"id": "user-123", "userId": "user-123", "name": "Aryan"})

Logger

Initialize a structured logger with OTLP export support.

from corekit.logger import LoggerService

service = LoggerService(
    service_name="my-service",
    log_level="INFO",
    enable_console=True
)
logger = service.get_logger()

logger.info("Service started", {"env": "production"})

Middleware

Add middleware to your FastAPI or Starlette app.

from starlette.applications import Starlette
from corekit.middleware import RateLimiterMiddleware

app = Starlette()
app.add_middleware(RateLimiterMiddleware, limit=100, window=60)

PostgreSQL

High-performance async PostgreSQL access.

from corekit.postgres import create_pg_pool, PgRepo

# Initialize connection pool
pool = await create_pg_pool(dsn="postgres://user:pass@host/db")

# Use Generic Repository
repo = PgRepo(pool, "users")
user = await repo.get_item(1)

# Custom Query
rows = await repo.run_query("SELECT * FROM users WHERE active = $1", True)

Redis Cache

from corekit.redis_cache import RedisClientService, RedisCache

client = RedisClientService(host="localhost", port=6379)
cache = RedisCache(client.get_client())

await cache.set("key", {"foo": "bar"}, expiry_in_seconds=60)
value = await cache.get("key")

Utils

Access standardized utility functions.

from corekit.utils import sleep, get_distance_from_lat_lon_in_km, encrypt_key

# Async Sleep
await sleep(1000) # 1 second

# Geo
dist = get_distance_from_lat_lon_in_km(40.7128, -74.0060, 34.0522, -118.2437)

# Encryption
encrypted = encrypt_key("my-secret", security_key="master-key")

Configuration

Most modules can be configured via constructor arguments. For seamless integration, it is recommended to use pydantic-settings or environment variables to inject credentials securely.

Common Environment Variables used implicitly by some underlying SDKs:

• AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, AZURE_TENANT_ID (for Azure Identity)
• OTEL_EXPORTER_OTLP_ENDPOINT (for OpenTelemetry)

Contributing

1. Clone the repository.
2. Create settings for your local dependencies (Redis, Postgres, etc.) or use mocks.
3. Install dependencies: pip install -e ".[dev]"
4. Run tests with pytest.

License

MIT