semvec/pypi-packages huggingface dataset:
os-py-lib-eb 1.2.0
Shared Redis, MongoDB, and healthcheck runtime helpers for EB services
License: BSD
Requires Python: >=3.10
Keywords: redis, mongodb, motor, healthcheck, runtime

# 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

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

### Development

```bash
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

```python
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:

```python
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"),
)
```