c1groupy 0.5.3

C1G company Python module for utilities and project setup

c1gpy

C1G company Python module for utilities and project setup.

Installation

Install the package using pip or uv:

pip install c1groupy

Or with uv:

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

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:

initialize-streamlit-project my-streamlit-app --author-email [e-mail]

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

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

Option 2: Docker Development

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

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

Option 3: Run Tests

Locally:

uv run pytest tests/

In Docker:

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:

import streamlit as st

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

3. Register the page in src/app/App.py:

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

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:

initialize-fastapi-project my-api --author-email [e-mail]

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

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

Option 2: Docker Development

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:

uv run pytest tests/

In Docker:

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:

async def get_data() -> dict:
    return {"data": "example"}

2. Add route in src/routers/router_name.py:

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:

# 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.

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:

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.

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:

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.

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.

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

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

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

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

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

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

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

# 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

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.

# 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

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:

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.

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:

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

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

Batch

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

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

Streaming

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:

from c1gpy.llm_handler import ModelProvider, LLMHandler

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

License

MIT