os-py-lib-eb 1.2.0

Shared Redis, MongoDB, and healthcheck runtime helpers for EB services

OS.py EB Runtime Library

Shared runtime helpers for EB services:

• Redis client setup and reconnect wrappers
• MongoDB client setup and reconnect wrappers
• Reusable healthcheck servers (async TCP and threaded HTTP)
• Unified console logging formatter

Author: Oleg Smirnov

Installation

From GitHub

pip install git+https://github.com/BestianCode/os.py.lib.eb.git

Development

git clone https://github.com/BestianCode/os.py.lib.eb.git
cd os.py.lib.eb
pip install -e .

Modules

os_eb.os_redis

• validate_redis_url(redis_url)
• build_async_redis(redis_url, ...)
• reconnect_async_redis(current_client, redis_url, ...)
• call_with_reconnect(get_client, set_client, redis_url, op_name, *args, **kwargs)

os_eb.os_mongo

• is_mongo_connection_error(exc)
• build_async_mongo_client(mongodb_uri, **kwargs)
• reconnect_async_mongo(current_client, mongodb_uri, **kwargs)
• build_sync_mongo_client(mongodb_uri, **kwargs)
• reconnect_sync_mongo(current_client, mongodb_uri, **kwargs)

os_eb.os_health

• parse_cidrs(raw)
• is_ip_allowed(ip_raw, allowed_networks)
• AsyncHealthServer for async workers
• ThreadedHealthServer for sync workers
  • configurable paths: health_path, dbhealth_path

os_eb.os_logging

• configure_console_logging(level="INFO", force=True)
• output format:
  • YYYY-MM-DD HH:MM:SS.mmm - LEVEL - HOSTNAME - message

Example

from os_eb.os_redis import build_async_redis, call_with_reconnect

redis = build_async_redis("redis://localhost:6379/0")

async def get_client():
    return redis

async def set_client(new_client):
    global redis
    redis = new_client

await call_with_reconnect(
    get_client=get_client,
    set_client=set_client,
    redis_url="redis://localhost:6379/0",
    op_name="xadd",
    op_args=("stream:name", {"k": "v"}),
)

Health paths from env:

import os
from os_eb.os_health import AsyncHealthServer

server = AsyncHealthServer(
    bind_host="0.0.0.0",
    port=8080,
    allowed_networks=[],
    dbhealth_provider=my_dbhealth_provider,
    health_path=os.getenv("HEALTHCHECK_PATH", "/alive"),
    dbhealth_path=os.getenv("DBHEALTHCHECK_PATH", "/dlive"),
)