c1groupy 0.5.3
C1G company Python module for utilities and project setup
Requires Python: <4.0,>=3.12

# c1gpy
[![Publish to PyPI](https://github.com/Career1Group/c1gpy/actions/workflows/publish_package.yaml/badge.svg)](https://github.com/Career1Group/c1gpy/actions/workflows/publish_package.yaml)

C1G company Python module for utilities and project setup.

## Installation

Install the package using pip or uv:

```bash
pip install c1groupy
```

Or with uv:

```bash
uv add c1groupy
```

### Installation Options (Extras)

You can install optional add-ins for specific functionality:

| Option | Command | Description |
|--------|---------|-------------|
| **PDF** | `pip install c1groupy[pdf]` | Installs `reportlab` for PDF generation. |
| **Utils** | `pip install c1groupy[utils]` | Installs `tqdm` and `cachetools`. |
| **All** | `pip install c1groupy[all]` | Installs all optional dependencies. |

## Features

### 1. Streamlit Project Initialization

The `initialize-streamlit-project` command creates a fully configured Streamlit multi-page application with Docker support and Pre-commit hooks.

#### Usage

```bash
initialize-streamlit-project <project-name> [--path <directory>] [--author-email <email>]
```

**Arguments:**
- `project-name` (required): Name of the project to create
- `--path` (optional): Directory where the project should be created (default: current directory)
- `--author-email` (optional): Author email address for project metadata

**Example:**

```bash
initialize-streamlit-project my-streamlit-app --author-email developer@example.com
```

This will create a new directory `my-streamlit-app/` with the following structure:

```
my-streamlit-app/
├── src/
│   ├── app/                # Streamlit application
│   │   ├── App.py          # Main entry point
│   │   └── pages/          # Additional pages
│   │       ├── 1_Page_1.py
│   │       └── 2_Page_2.py
│   └── entrypoint.sh       # Container entrypoint script
├── tests/                  # Test files
│   └── test_example.py
├── Dockerfile              # Docker configuration
├── docker-compose.yml      # Docker Compose configuration
├── pyproject.toml          # Project dependencies
├── .pre-commit-config.yaml # Pre-commit hooks
├── .gitignore              # Git ignore rules
└── README.md               # Project documentation
```

#### What's Included

**Multi-page Streamlit App:**
- Main page (`App.py`) with welcome content
- Two example pages demonstrating charts and forms
- Proper page configuration and navigation

**Docker Support:**
- `Dockerfile` with Python 3.12 and uv package manager
- `docker-compose.yml` configured for local development
- Smart entrypoint script that supports:
  - `LOCALRUN=TRUE`: Hot-reloading for development
  - `LOCALRUN=FALSE`: Production mode
  - `PYTEST=TRUE`: Run tests in container
- Volume mounting of `src/` directory for live code updates

**Development Tools:**
- `.pre-commit-config.yaml` with ruff linting and formatting
- `.gitignore` configured for Python projects
- Git repository initialized with initial commit
- Example test file with pytest

**Documentation:**
- Comprehensive README with setup and usage instructions

#### Getting Started with Your New Project

After creating a project, navigate to it and choose your development method:

**Option 1: Local Development with uv**

```bash
uv sync
uv run streamlit run src/app/App.py
```

**Option 2: Docker Development**

```bash
docker-compose up --build --force-recreate --remove-orphans
```

The application will be available at http://localhost:8501

**Option 3: Run Tests**

Locally:
```bash
uv run pytest tests/
```

In Docker:
```bash
docker-compose run -e PYTEST=TRUE streamlit-app
```

#### Adding New Pages

This template uses Streamlit's `st.navigation()` for organized multi-page navigation.

To add a new page:

1. Create a new Python file in `src/app/pages/` (e.g., `analytics.py`)
2. Add your page content:
```python
import streamlit as st

st.title("Analytics Dashboard")
# Your page content here
```

3. Register the page in `src/app/App.py`:
```python
pages = {
    "📊 Main": [
        st.Page("pages/1_Page_1.py", title="Page 1", icon="📈"),
        st.Page("pages/analytics.py", title="Analytics", icon="📊"),  # New page
    ],
    # ...
}
```

#### Docker Environment Variables

The entrypoint script supports the following environment variables:

- `LOCALRUN`: Set to `TRUE` for development mode with hot-reloading (default: `FALSE`)
- `PYTEST`: Set to `TRUE` to run tests instead of the app (default: `FALSE`)
- `PORT`: Port for Streamlit server (default: `8501`)

---

## 2. FastAPI Project Initialization

The initialize-fastapi-project command creates a fully configured FastAPI application with organized router structure, Docker support, and Pre-commit hooks.

### Usage

```bash
initialize-fastapi-project <project-name> [--path <directory>] [--author-email <email>]
```

**Arguments:**
- `project-name` (required): Name of the project to create
- `--path` (optional): Directory where the project should be created (default: current directory)
- `--author-email` (optional): Author email address for project metadata

**Example:**

```bash
initialize-fastapi-project my-api --author-email developer@example.com
```

This will create a new directory `my-api/` with the following structure:

```
my-api/
├── src/
│   ├── api_interface.py    # Main FastAPI app
│   ├── routers/            # API routers
│   │   ├── router1.py
│   │   └── router2.py
│   ├── endpoints/          # Endpoint logic
│   │   ├── router1/
│   │   │   └── example_endpoint.py
│   │   └── router2/
│   │       └── example_endpoint.py
│   └── entrypoint.sh       # Container entrypoint script
├── tests/                  # Test files
│   └── test_example.py
├── Dockerfile              # Docker configuration
├── docker-compose.yml      # Docker Compose configuration
├── pyproject.toml          # Project dependencies
├── .pre-commit-config.yaml # Pre-commit hooks
├── .gitignore              # Git ignore rules
└── README.md               # Project documentation
```

### What's Included

**FastAPI Application:**
- Main API interface with CORS middleware
- Two example routers with organized structure
- Async endpoint examples
- API key authentication (commented out, ready to enable)
- Automatic interactive documentation (Swagger UI & ReDoc)

**Docker Support:**
- Multi-stage Dockerfile with Python 3.12 and uv
- `docker-compose.yml` configured for local development
- Hypercorn ASGI server with uvloop for performance
- Smart entrypoint script that supports:
  - `LOCALRUN=TRUE`: Hot-reloading for development
  - `LOCALRUN=FALSE`: Production mode with 4 workers
  - `PYTEST=TRUE`: Run tests in container
- Volume mounting of `src/` directory for live code updates

**Development Tools:**
- `.pre-commit-config.yaml` with ruff linting and formatting
- `.gitignore` configured for Python projects
- Git repository initialized with initial commit
- Example test file with pytest and TestClient

**Documentation:**
- Comprehensive README with API structure explanation
- Examples for adding new endpoints and routers

### Getting Started with Your New FastAPI Project

After creating a project, navigate to it and choose your development method:

**Option 1: Local Development with uv**

```bash
uv sync
uv run hypercorn src.api_interface:rest_api --bind :8000 --reload
```

**Option 2: Docker Development**

```bash
docker-compose up --build --force-recreate --remove-orphans
```

The API will be available at:
- **API**: http://localhost:8000
- **Interactive Docs**: http://localhost:8000/docs
- **ReDoc**: http://localhost:8000/redoc

**Option 3: Run Tests**

Locally:
```bash
uv run pytest tests/
```

In Docker:
```bash
docker-compose run -e PYTEST=TRUE fastapi-app
```

### API Structure

The FastAPI template uses a clean separation of concerns:

- **`api_interface.py`**: Main FastAPI app with middleware and router registration
- **`routers/`**: Route definitions and request/response handling
- **`endpoints/`**: Business logic separated by router

### Adding New Endpoints

1. Create endpoint logic in `src/endpoints/router_name/new_endpoint.py`:
```python
async def get_data() -> dict:
    return {"data": "example"}
```

2. Add route in `src/routers/router_name.py`:
```python
from endpoints.router_name import new_endpoint

@router.get("/data")
async def get_data():
    return await new_endpoint.get_data()
```

### API Security

The template includes commented-out API key authentication that can be easily enabled:

```python
# In routers/router1.py
from fastapi.security import APIKeyHeader
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)

@router.get("/secure")
async def secure_endpoint(api_key: str = Security(api_key_header)):
    # Validate api_key
    return {"secure": "data"}
```

### Docker Environment Variables

- `PORT`: Port for the API server (default: `8000`)
- `LOCALRUN`: Set to `TRUE` for development mode with hot-reloading (default: `FALSE`)
- `PYTEST`: Set to `TRUE` to run tests instead of the app (default: `FALSE`)

---

## 3. Logging

Reusable logging utilities with colorized console output and Google Cloud Logging support.

### Local Development Logger

Pretty-printed console output with colors using Rich, plus optional file logging.

```python
from c1gpy.logging import get_logger

# Basic usage
logger = get_logger(__name__, level="DEBUG")
logger.info("Application started")
logger.debug("Debug information")
logger.error("Something went wrong")

# With file logging
logger = get_logger(
    __name__,
    level="INFO",
    log_dir="./logs",
    log_filename="app.log"
)
```

**Fluent builder pattern:**

```python
from c1gpy.logging import C1GLogger

logger = (
    C1GLogger("my_app")
    .with_level("DEBUG")
    .with_file_logging("./logs", filename="app.log")
    .build()
)
```

**Parameters:**
- `name`: Logger name (typically `__name__`)
- `level`: Log level (`DEBUG`, `INFO`, `WARNING`, `ERROR`, `CRITICAL`)
- `console_format`: Format string for console output
- `file_format`: Format string for file output
- `log_dir`: Directory for log files (enables file logging)
- `log_filename`: Custom log file name

### Google Cloud Logger

For production deployments on GCP. Sends logs to Google Cloud Logging without console output.

```python
from c1gpy.logging import get_cloud_logger

logger = get_cloud_logger(__name__, level="INFO")
logger.info("Application started")
logger.error("Something went wrong")  # Visible in GCP Error Reporting
```

**Fluent builder pattern:**

```python
from c1gpy.logging import C1GCloudLogger

logger = (
    C1GCloudLogger("my_app")
    .with_level("DEBUG")
    .build()
)
```

**Requirements:**
- `google-cloud-logging` package (included in dependencies)
- GCP authentication (Application Default Credentials, service account, etc.)

**Parameters:**
- `name`: Logger name (typically `__name__`)
- `level`: Log level (`DEBUG`, `INFO`, `WARNING`, `ERROR`, `CRITICAL`)
- `log_format`: Format string for log messages

---

## 4. Utilities

Common utility functions for authentication and HTTP requests.

### Password Hashing (Argon2)

Secure password hashing using Argon2id algorithm.

```python
from c1gpy.utils import hash_password, verify_password

# Hash password for storage
hashed = hash_password("user_password")

# Verify on login
if verify_password(input_password, stored_hash):
    print("Login successful")
```

### Async HTTP Client

HTTP client with retry, rate limiting, exponential backoff, and HTTP/2 support.

```python
from c1gpy.utils import AsyncHTTPClient, HTTPClientError, JSONDecodeError

# Basic usage (no retries, HTTP/2 enabled)
async with AsyncHTTPClient() as client:
    data = await client.get("https://api.example.com/data")

# With retries and backoff
client = AsyncHTTPClient(
    base_url="https://api.example.com",
    retries=3,
    backoff_factor=2.0,      # delays: 2s, 4s, 8s
    rate_limit_delay=0.5,    # min 0.5s between requests
    http2=True,              # default
)

try:
    data = await client.get("/endpoint")
    data = await client.post("/create", json={"key": "value"})
except HTTPClientError as e:
    print(f"Request failed: {e}, status: {e.status_code}")
except JSONDecodeError as e:
    print(f"Invalid JSON response: {e}")
finally:
    await client.close()
```

**Parameters:**
- `base_url`: Optional base URL for all requests
- `retries`: Number of retry attempts (default: 0)
- `backoff_factor`: Multiplier for exponential backoff (default: 1.0)
- `rate_limit_delay`: Minimum delay between requests in seconds (default: 0)
- `timeout`: Request timeout in seconds (default: 30)
- `http2`: Use HTTP/2 protocol (default: True)

**Methods:** `get`, `post`, `put`, `patch`, `delete`

**Exceptions:**
- `HTTPClientError`: Raised when request fails after all retries (includes `status_code`)
- `JSONDecodeError`: Raised when response is not a valid JSON dict

---

## 5. Google Cloud Utilities

Clients for Google Cloud services. All clients support service account JSON credentials or Application Default Credentials.

### Secret Manager

```python
from c1gpy.google_utils import SecretManagerClient

client = SecretManagerClient("my-project")

# Get a secret
api_key = client.get_secret("api-key")
db_password = client.get_secret("db-password", version="2")

# Create a new secret
client.create_secret("new-secret", "secret-value")

# List all secrets
secrets = client.list_secrets()
```

### Google Sheets

```python
from c1gpy.google_utils import GoogleSheetsClient

client = GoogleSheetsClient("service-account.json")

# Read data
data = client.read_sheet("spreadsheet_id", "Sheet1!A1:D10")

# Write data
client.write_sheet("spreadsheet_id", "Sheet1!A1", [["Name", "Age"], ["Alice", 30]])

# Append data
client.append_sheet("spreadsheet_id", "Sheet1!A1", [["Bob", 25]])

# Clear a range
client.clear_sheet("spreadsheet_id", "Sheet1!A1:D10")
```

### Cloud Storage

```python
from c1gpy.google_utils import CloudStorageClient

client = CloudStorageClient()

# Upload data
client.upload_blob("my-bucket", "data.json", '{"key": "value"}')
client.upload_file("my-bucket", "image.png", "/path/to/image.png")

# Download data
content = client.download_blob("my-bucket", "data.json")
client.download_blob_to_file("my-bucket", "data.json", "/local/path.json")

# List and check blobs
blobs = client.list_blobs("my-bucket", prefix="data/")
exists = client.blob_exists("my-bucket", "data.json")

# Delete
client.delete_blob("my-bucket", "data.json")
```

### Google Drive

```python
from c1gpy.google_utils import GoogleDriveClient

client = GoogleDriveClient("service-account.json")

# List files
files = client.list_files()
files = client.list_files(folder_id="folder_id")

# Upload
result = client.upload_file("/path/to/file.pdf", name="document.pdf", folder_id="folder_id")
result = client.upload_bytes(b"content", "file.txt")

# Download
content = client.download_file("file_id")
client.download_file_to_path("file_id", "/local/path.pdf")

# Create folder
folder = client.create_folder("New Folder", parent_folder_id="parent_id")

# Delete
client.delete_file("file_id")
```

### Google Docs

```python
from c1gpy.google_utils import GoogleDocsClient

client = GoogleDocsClient("service-account.json")

# Read document
doc = client.get_document("document_id")
text = client.get_document_text("document_id")

# Create document
new_doc = client.create_document("My Document")

# Edit document
client.insert_text("document_id", "Hello, World!", index=1)
client.replace_text("document_id", "old text", "new text")
client.delete_content("document_id", start_index=1, end_index=10)
```

**Requirements:**
- GCP authentication (service account JSON or Application Default Credentials)
- Appropriate API permissions enabled in GCP project

---

## 6. Payload Explorer

Read-only client for exploring and exporting data from the Payload CMS. Features TTL-based caching, progress reporting, and multiple export formats.

### Basic Usage

```python
from c1gpy.payload_explorer import PayloadExplorer

# Initialize with environment and API key
explorer = PayloadExplorer(
    environment="PROD",  # or "STAGE"
    api_key="your-api-key",
)

# Fetch by Content ID (e.g., "ku1", "ku1_mo1_le1_ak1_b")
course = explorer.get_course_by_content_id("ku1")
activity = explorer.get_activity_by_content_id("ku1_mo1_le1_ak1_b")

# Fetch by Payload ID (MongoDB ObjectId)
course = explorer.get_course_by_id("67a1b2c3d4e5f6...")

# Fetch by name
course = explorer.get_course_by_name("Einführung in Python")
```

### Available Methods

Each entity type (Course, Module, LearningUnit, Activity) supports three fetch methods:

| Method | Description | Example Input |
|--------|-------------|---------------|
| `get_X_by_id(payload_id)` | Fetch by Payload ID | `"67a1b2c3d4e5f6..."` |
| `get_X_by_content_id(content_id)` | Fetch by Content ID | `"ku1"`, `"ku1_mo1_le1_ak1_b"` |
| `get_X_by_name(name)` | Fetch by name | `"Kursname"` |

Replace `X` with `course`, `module`, `learning_unit`, or `activity`.

### Batch Operations

```python
# Get all activities for a course
activities = explorer.get_activities_for_course("ku1")

# Filter by activity type: "b" (ebook), "q" (quiz), "a" (assessment)
ebooks = explorer.get_activities_for_course("ku1", activity_type="b")

# Get overview of all entities
courses = explorer.get_overview("courses", limit=50)
modules = explorer.get_overview("modules", limit=100)
```

### Export Functions

```python
from pathlib import Path

# Export course hierarchy to CSV
explorer.export_course_overview_csv(
    course_id="ku1",
    output_path=Path("./output/ku1_overview.csv"),
    include_modules=True,
    include_learning_units=True,
    include_activities=True,
)

# Export all ebooks in a course as PDFs
# Prompts for confirmation if more than 20 PDFs
created_files = explorer.export_pdfs_for_course(
    course_id="ku1",
    output_dir=Path("./output/pdfs/"),
    confirm_threshold=20,
)

# Export a single activity as PDF
explorer.export_activity_as_pdf(
    activity_content_id="ku1_mo1_le1_ak1_b",
    output_path=Path("./output/ebook.pdf"),
)
```

### Caching

All fetch operations are cached with TTL (Time-To-Live). Default is 300 seconds.

```python
# Custom cache TTL (10 minutes)
explorer = PayloadExplorer(
    environment="PROD",
    api_key="your-api-key",
    cache_ttl_seconds=600,
)

# Clear cache manually
explorer.clear_cache()
```

### Configuration Parameters

| Parameter | Default | Description |
|-----------|---------|-------------|
| `environment` | required | `"PROD"` or `"STAGE"` |
| `api_key` | required | Payload CMS API key |
| `cache_ttl_seconds` | `0` | Cache lifetime in seconds |
| `request_delay` | `0.3` | Delay between API requests (rate limiting) |

### Exceptions

```python
from c1gpy.payload_explorer import (
    PayloadExplorerError,    # Base exception
    AuthenticationError,     # Invalid API key
    ResourceNotFoundError,   # Entity not found
    RateLimitError,          # API rate limit exceeded
    ExportError,             # Export operation failed
)

try:
    course = explorer.get_course_by_content_id("invalid")
except AuthenticationError as e:
    print(f"Auth failed: {e}")
except ResourceNotFoundError as e:
    print(f"Not found: {e.identifier}")
```

### Data Models

All fetched entities are returned as Pydantic models:

```python
course = explorer.get_course_by_content_id("ku1")

print(course.id)          # Payload ID
print(course.content_id)  # Content ID (e.g., "ku1")
print(course.name)        # Course name
print(course.modules)     # List of module IDs
print(course.raw_data)    # Original API response dict
```

---

## 7. LLM Handler

Unified interface for calling LLMs (OpenAI, Anthropic, Google Gemini) with:

- **Langfuse** prompt management (prompt fetch + compilation)
- **Tracing** via Langfuse callback handler
- **Sync + async** calls
- **Batch** calls
- **Streaming** responses

### Basic usage

The `LLMHandler` requires an initialized `Langfuse` client. It automatically infers the model name, temperature, and max tokens from the Langfuse prompt configuration if not explicitly provided.

```python
from langfuse import Langfuse
from c1gpy.llm_handler import LLMHandler

# Initialize Langfuse client (uses environment variables: LANGFUSE_SECRET_KEY, LANGFUSE_PUBLIC_KEY, LANGFUSE_HOST)
langfuse = Langfuse()

# Initialize handler
# model_name, temperature, etc. are inferred from the prompt config
handler = LLMHandler(
    langfuse,
    "my-prompt",
    api_key="sk-...", # Optional if set in environment
)

result = handler.call_model(keyword="test")
```

### Initialization Parameters

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `langfuse` | `Langfuse` | (required) | Initialized Langfuse client instance. |
| `langfuse_prompt_name` | `str` | (required) | Name of the prompt in Langfuse. |
| `model_name` | `str` | `None` | Model to use (e.g., `gpt-4o`). Infers from Langfuse prompt config if `None`. |
| `api_key` | `str` | `None` | API key for the provider. Optional if set in environment. |
| `langfuse_prompt_label` | `str` | `"production"` | Label for the Langfuse prompt version. |
| `provider` | `ModelProvider` | `None` | Explicit provider override. Auto-detected from `model_name` if `None`. |
| `response_model` | `type[T]` | `None` | Pydantic model for structured output. Mutually exclusive with `json_mode`. |
| `json_mode` | `bool` | `False` | If `True`, returns unvalidated JSON dict. Mutually exclusive with `response_model`. |
| `temperature` | `float` | `None` | Sampling temperature. Infers from Langfuse prompt config or defaults to `1.0`. |
| `max_tokens` | `int` | `None` | Maximum tokens in response. Infers from Langfuse prompt config if `None`. |
| `timeout` | `float` | `None` | Request timeout in seconds. |
| `cache_ttl_seconds` | `int` | `0` | TTL for Langfuse prompt caching in seconds. |
| `retry_config` | `dict` | `None` | Configuration for LangChain retries (e.g., `{"max_retries": 2}`). |

### Advanced initialization

You can override inferred parameters by passing them explicitly:

```python
handler = LLMHandler(
    langfuse,
    "my-prompt",
    model_name="gpt-4o",        # Override model
    temperature=0.7,            # Override temperature
    json_mode=True,             # Enable JSON mode
    retry_config={"max_retries": 2} # Configure retries
)
```

### Async

```python
result = await handler.acall_model(keyword="test")
```

### Batch

```python
results = handler.batch_invoke(
    [{"keyword": "a"}, {"keyword": "b"}, {"keyword": "c"}]
)

results_async = await handler.abatch_invoke(
    [{"keyword": "a"}, {"keyword": "b"}, {"keyword": "c"}]
)
```

### Streaming

```python
for chunk in handler.call_model_stream(keyword="test"):
    print(chunk, end="")

async for chunk in handler.acall_model_stream(keyword="test"):
    print(chunk, end="")
```

### Provider selection

Providers are auto-detected from `model_name` prefixes (e.g. `gpt-*`, `claude-*`, `gemini-*`).
You can also explicitly override the provider:

```python
from c1gpy.llm_handler import ModelProvider, LLMHandler

handler = LLMHandler(
    langfuse,
    "my-prompt",
    model_name="gpt-4o",
    provider=ModelProvider.OPENAI,
)
```

## License

MIT