keble-helpers 1.20.0

No project description provided

Keble helpers

Just a collection of helper functions used by keble project.

Version 1.12.16 Aliyun Python 3.13 import compatibility

• Preserves the existing from keble_helpers import AliyunOss public export.
• Adds explicit Python 3.13 compatibility dependencies for Aliyun OSS imports: legacy-cgi, six, and the aliyun-python-sdk-core 2.16.x line.
• Bridges the Aliyun SDK's older vendored six.moves module paths at module import time, so downstream packages can import helper schemas without install-order-dependent failures.

Version 1.12.11 update

• Adds package-neutral UsageAccountingEvent, UsageAccountingRecorderProtocol, UsageAccountingSource, and UsageAccountingUnitType.
• The usage-accounting contract intentionally contains no MongoDB, task id, pricing, or backend business logic. Host services decide how to price and persist emitted events.
• Counted API/item/search events require unit_count; token events require Pydantic-AI RunUsage.

Shared Typings

keble-helpers owns package-neutral shared enums and value objects that need to stay stable across backend packages.

1. Marketplace
2. Language
3. CommerceEntityType

CommerceEntityType is the canonical cross-package enum for commerce entities:

1. BRAND_IDENTITY
2. BRAND_MENTION
3. CATEGORY
4. PRODUCT
5. LISTING
6. SKU

Display helpers on CommerceEntityType are part of the shared contract as well. upper_snake_to_title() must preserve known acronyms such as SKU instead of degrading them to title-cased words.

Version 1.12.10 update

• Versions the helper publication-blocker documentation on the maintained 1.12.x line.
• Runtime helper schemas and protocols are unchanged from 1.12.9.

Agent Runtime Context

keble-helpers 1.12.10 includes AgentBaseDeps, a package-neutral Pydantic deps base for browser-selected agent context.

1. marketplace carries the commerce marketplace scope when tools need it.
2. language carries the frontend language that prompts and user-visible agent output should respect.
3. Agent packages can combine it with database deps through multiple inheritance, for example class MyDeps(AgentDbDeps, AgentBaseDeps): ....

Do not expose internal datasource or provider names in user-facing bootstrap chat copy. Keep such names in provider schemas, internal logs, or tests only.

AgentBaseDeps Publication Status

AgentBaseDeps is built in keble-helpers 1.12.10, but this environment cannot publish it to the configured package index unless PyPI credentials or OIDC trusted-publishing token are available.

1. uv build should produce dist/keble_helpers-1.12.10-py3-none-any.whl.
2. python -m pip index versions keble-helpers still shows 1.12.1 as the latest visible index release.
3. If uv publish dist/keble_helpers-1.12.10-py3-none-any.whl fails with missing credentials, service repos must consume the bundled wheel until a credentialed publish is performed.
4. The 2026-05-23 local publish attempt failed for that credential reason, so this workspace still treats 1.12.10 as a built, bundled-wheel release.

ObjectId Boundary Rule

keble_helpers.ObjectId is a Pydantic annotation over BSON ObjectId, not a replacement for every Mongo query id. It serializes to a string when schemas are dumped with model_dump(mode="json").

Use keble_helpers.ObjectId in Pydantic schemas that cross API, chat, tool, or queue boundaries. Use direct BSON ids in Mongo query internals and tests when that is clearer, preferably aliased as BsonObjectId.

Queued Envelope Contract

keble-helpers owns queue-neutral envelope types that let backend and package clients share one broker contract without sharing Celery internals.

1. QueuedEnvelope is the persisted message body:
   • job_type tells processors whether they own the work,
   • payload is opaque to the dispatcher and validated by the handling processor,
   • domain_refs links optional domain objects such as task, grid, or positioning ids.
2. QueuedEnvelopeProcessContext wraps one envelope plus attempt/runtime metadata.
3. QueuedEnvelopeProcessResult tells the backend dispatcher whether a processor handled the envelope and whether handling succeeded.
4. QueuedEnvelopeProcessorProtocol is the shared client-side contract:

async def aprocess_queued_envelope(
    context: QueuedEnvelopeProcessContext,
) -> QueuedEnvelopeProcessResult:
    """Process owned job types and return handled=False for unrelated jobs."""

Backend owns Celery, queue ordering, retries, and persisted ledger status. Package clients own their job-type payload schemas and validation.

Agentic Action Events

keble-helpers owns the package-neutral event envelope used by task, positioning, segmenting, and backend action runtimes.

1. AgenticActionEvent[T] wraps one package-owned typed result payload.
2. AgenticActionEventSource is the single source of truth for the emitting package identity; source is an enum, not a free string, and keble-core mirrors these values 1:1 for the frontend.
3. AgenticActionEventStatus records lifecycle state such as SUCCEEDED or FAILED. It is now a backward-compatible alias of the unified AgenticActionStatus (see "Unified Agentic Action Contract" below).
4. AgenticActionEvent.started/progressed/succeeded/failed(...) are the canonical cls factories. Use them instead of hand-building the envelope so subclasses never inherit a wrong status default.
5. AgenticEventEmitter calls ordered async callbacks and propagates callback failures.
6. AgenticEventEmitter.build(...) lets callers pass None, callback sequences, or a ready emitter.

event = AgenticActionEvent[MyActionResult].succeeded(
    source=AgenticActionEventSource.KEBLE_SEGMENTING,
    action_type="UPDATE_DIMENSIONS",
    payload=result,
    root_id=str(grid_id),
)
await emitter.aemit(event)

The helper contract deliberately does not depend on Celery, FastAPI, or SSE. Backend can later bridge the same event envelope into listeners without each package inventing a new callback shape. Every agentic package (task, segmenting, positioning, amz-product-report) sets source from AgenticActionEventSource so the backend Redis publisher and the single task-room WebSocket carry one consistent envelope.

Unified Agentic Action Contract

Raising a browser client tool, asking for a server-tool approval, and self-serving a subagent decision are all the same kind of thing — an action that pauses or reports on an agent run. They therefore share one canonical contract in keble_helpers/ai/client_action.py, instead of each host re-defining its own status/progress/kind enums (the former keble_agentic_chat.ChatActionStatus / ChatActionProgress / ChatActionKind are now removed in favor of these).

1. AgenticActionStatus is the single superset lifecycle enum. It carries both the event-lifecycle members (STARTED/PROGRESSED/SUCCEEDED/FAILED) and the resolution members (PENDING/APPROVED/DENIED/SUBMITTED/REJECTED/ABANDONED). AgenticActionEventStatus is now a backward-compatible alias of this one enum, so event runtimes and action runtimes never disagree on a status string.
2. AgenticActionKind enumerates the four families: SERVER_TOOL_APPROVAL, CLIENT_TOOL, SERVER_SELF_SERVED (a background subagent answering its own client tool with the most-likely user choice, no human in the loop), and SERVER_TOOL_CALL (an auto-executed backend tool call recorded for the visible timeline — created directly terminal SUCCEEDED/FAILED, never pending, never resolvable).
3. AgenticActionProgress is the canonical progress shape (the renamed home of the former ChatActionProgress), with .queued(...) and .to_action_status().
4. AgenticClientActionBase is the package-neutral base every host action inherits (keble_agentic_chat.ChatAction inherits it). It owns identity (action_id, tool_call_id, tool_name), kind, optional source (AgenticActionEventSource), status, JSON-safe request/result, optional progress, and timestamps.

Rule for all packages: any action that pauses or reports an agent run inherits AgenticClientActionBase and uses AgenticActionStatus/AgenticActionProgress. No package re-defines a parallel status or progress enum.

from keble_helpers import (
    AgenticActionKind, AgenticActionStatus, AgenticClientActionBase, AgenticActionEventSource,
)

class ChatAction(AgenticClientActionBase, ChatValueBase):
    """A host action inherits the canonical contract and only adds serialization policy."""

action = ChatAction(
    action_id=tool_call_id, kind=AgenticActionKind.CLIENT_TOOL,
    source=AgenticActionEventSource.KEBLE_AGENTIC_CHAT,
    tool_call_id=tool_call_id, tool_name="request_client_tool",
    status=AgenticActionStatus.PENDING, request={"tool_type": "MARKETPLACE_SELECT"},
    occurred_at=datetime.now(timezone.utc),
)

Agentic Chat Scope Runtime (cross-repo standard)

ChatScopeRuntimeProtocol (keble_helpers/ai/chat_scope.py) is the framework-neutral contract for "a per-scope agentic-chat runtime holder" — the object that owns one chat scope's durable history store plus the cooperative run-control used to stop an in-flight streaming run. The concrete implementation is keble_agentic_chat.AgenticChat (it binds a LangGraph turn engine and adds a per-turn runtime factory). The contract is declared HERE, in framework-neutral keble-helpers (no LangGraph / pydantic-ai types — store and chat_id are Any), so any consumer package can type against it without adding keble-agentic-chat as a dependency.

Standard for future agents: when a package or service needs to host an agentic-chat surface, type against ChatScopeRuntimeProtocol and reuse the package AgenticChat instead of re-inventing store/interrupt ownership inside a service repo.

from keble_helpers import ChatScopeRuntimeProtocol

async def stop_run(scope: ChatScopeRuntimeProtocol, *, owner: str, scope_id: str, chat_id: str) -> bool:
    """Type against the neutral contract; no keble-agentic-chat import needed."""
    return await scope.arequest_interrupt(
        owner=owner, scope_type="TASK", scope_id=scope_id, chat_id=chat_id, reason="user requested stop",
    )

Chat tool providers (cross-repo standard)

ChatToolProviderProtocol (keble_helpers/ai/chat_tool_provider.py, since 1.15.0) is the companion contract for "a domain that contributes tools to a chat scope". Instead of a host hardcoding every domain's register_*_tools call inline in a giant per-scope builder, each domain is wrapped as a small provider exposing provider_id + register(*, agent, context); the host composes a list of them with the canonical handler keble_agentic_chat.compose_tool_providers. The contract is framework-neutral (agent and context are Any), so a provider adapter need not import keble-agentic-chat or pydantic-ai.

Standard for future agents: to give a chat scope a new domain's tools, add a provider that satisfies ChatToolProviderProtocol and append it to the scope's provider list — do not edit the scope builder's internals or genericize the per-scope deps. Deps/data shapes stay explicit per scope (they diverge too much to share); only tool composition is unified here.

from typing import Any

from keble_helpers import ChatToolProviderProtocol


class MyDomainToolProvider:
    """Wrap an existing `register_my_tools` registrar as a chat tool provider."""

    provider_id = "my_domain"

    def __init__(self, *, client: Any) -> None:
        self._client = client

    def register(self, *, agent: Any, context: Any) -> None:
        register_my_tools(agent, client=self._client)  # existing domain registrar


_: ChatToolProviderProtocol = MyDomainToolProvider(client=object())

Chat memory contracts (cross-repo standard)

ChatMemoryRecord, ChatMemoryKind, and ChatMemoryStoreProtocol (keble_helpers/ai/chat_memory.py, since 1.16.0) are the framework-neutral durable-memory seam for agentic chats. The chat engine (keble_agentic_chat.LangGraphChatRuntime) recalls records before a turn and remembers new ones after it, but the record shape and store protocol live here so domain packages and hosts produce/consume the same records without importing the engine. kind stays a plain str for host-defined kinds; well-known values (EPISODE/FACT/PREFERENCE/SUMMARY) come from ChatMemoryKind, and the per-turn episodic record is built via ChatMemoryRecord.episode(...).

Scoping contract (locked by design): recall is shared across ALL chats of the same (owner, scope_type) pair — durable memory intentionally crosses chat sessions. scope_id/chat_id are write-side provenance metadata a store persists for diagnostics, NOT recall filters.

from keble_helpers import ChatMemoryKind, ChatMemoryRecord, ChatMemoryStoreProtocol


class MyVectorMemoryStore:
    """Host store over any backend (Qdrant, SQL, files...)."""

    async def arecall(self, *, owner, scope_type, scope_id, chat_id, query, limit=8):
        ...  # filter by owner + scope_type only (cross-chat recall by design)

    async def aremember(self, *, owner, scope_type, scope_id, chat_id, records):
        ...  # persist records; scope_id/chat_id stored as provenance metadata


_: ChatMemoryStoreProtocol = MyVectorMemoryStore()

Image Prompt Runtime

keble-helpers 1.12.9 owns the shared image prompt preflight, image-count budget, and provider fallback policy used by backend AI clients.

1. ImagePromptChecker accepts only model-supported image responses: image/png, image/jpeg, image/gif, and image/webp.
2. HTTP probes reject non-200/206 responses, unsupported response Content-Type values such as image/svg+xml, and extension-only URLs outside .png, .jpg, .jpeg, .gif, or .webp.
3. ImagePromptChecker.max_images_per_prompt keeps typed ImageUrl parts below the model/provider limit before the call. Non-image prompt parts are always preserved.
4. arun_with_image_fallback(...) now treats provider invalid_image_format, unsupported-image, and too-many-images errors like inaccessible image URLs: it retries once with image parts stripped, then preserves the original exception if the text-only retry still fails.
5. Backend services should keep using the shared checker instead of adding service-local image validation.

Agentic Tool Config

keble-helpers owns the shared pydantic-ai tool registration config used by package registrars. Packages should import AgentToolConfig rather than redefining local tool-name/description/approval schemas.

from keble_helpers import AgentToolConfig

config = AgentToolConfig.build(
    {
        "name": "mutate_segmenting",
        "description": "Apply one typed segmenting action batch.",
        "requires_approval": True,
    }
)

This helper only describes tool metadata. Package registrars still own the domain payload type and tool execution body.

Aliyun

The Aliyun module provides helpers for interacting with Alibaba Cloud (Aliyun) services.

Base Classes

Aliyun

• __init__(*, access_key: str, secret: str): Initialize with Aliyun credentials

OSS (Object Storage Service)

AliyunOss

• __init__(oss_endpoint: AnyHttpUrl, bucket: str, **kwargs): Initialize OSS client
• get_bucket() -> oss2.Bucket: Get OSS bucket instance
• get_bucket_with_sts(sts_token: str): Get bucket with STS token
• get_object_meta(key: str) -> AliyunOssMeta: Get object metadata
• save_object_to_local(key: str, local_path: str, *args, **kwargs): Download file from OSS
• save_local_to_cloud(key: str, local_path: str, *args, **kwargs): Upload file to OSS
• save_snapshot_to_local(key: str, local_path: str, seconds: int): Get video snapshot
• cold_archive_object(key: str): Convert object to cold archive storage class
• get_sts_signed_url(sts_token: str, key: str, *, expire_seconds: int = 60, content_type: Optional[str] = None, oss_storage_class: Optional[str] = None) -> str: Generate signed URL with STS

STS (Security Token Service)

AliyunSts

• __init__(region, **kwargs): Initialize STS client
• get_sts(session_name: str, role_arn: str) -> AliyunStsToken: Get STS token

Schemas

AliyunOssPutObjectResponse

• status: int: Response status
• request_id: str: Request ID
• etag: str: ETag
• headers: dict: Response headers

AliyunStsToken

• access_key_secret: str: Access key secret
• security_token: str: Security token
• access_key_id: str: Access key ID

AliyunOssMeta

• etag: Optional[str]: OSS ETag
• content_length: Optional[int]: File size in bytes
• last_modified: Optional[int]: Last modified timestamp
• content_type: Optional[str]: MIME type of the file

Usage Examples

from keble_helpers import AliyunOss

# Initialize Aliyun OSS
oss_client = AliyunOss(
    oss_endpoint="https://oss-cn-beijing.aliyuncs.com",
    bucket="your-bucket-name",
    access_key="your-access-key-id",
    secret="your-access-key-secret"
)

# Upload file to OSS
response = oss_client.save_local_to_cloud(
    key="path/in/oss/file.txt",
    local_path="/local/path/to/file.txt"
)

# Get file metadata
meta = oss_client.get_object_meta("path/in/oss/file.txt")

# Download file from OSS
oss_client.save_object_to_local(
    key="path/in/oss/file.txt",
    local_path="/local/path/to/download.txt"
)

# Get STS token
sts_client = AliyunSts(
    region="cn-beijing",
    access_key="your-access-key-id", 
    secret="your-access-key-secret"
)
sts_token = sts_client.get_sts(
    session_name="session-name",
    role_arn="acs:ram::your-account-id:role/your-role-name"
)

# Generate signed URL with STS token
signed_url = oss_client.get_sts_signed_url(
    sts_token=sts_token.security_token,
    key="path/in/oss/file.txt",
    expire_seconds=3600
)

Progress

The Progress module provides a Redis-based task tracking system to monitor the progress of multi-stage operations.

Base Classes

ProgressHandler

• __init__(redis: Redis): Initialize with Redis connection
• new(*, key: str, model_key: str | None = None) -> ProgressTask: Create a new progress task
• get(*, key: str) -> ProgressReport | None: Retrieve progress report by key

ProgressTask

• __init__(redis: Optional[Redis] = None, key: Optional[str] = None, model_key: Optional[str] = None, root: Optional["ProgressTask"] = None): Initialize a progress task
• new_subtask() -> ProgressTask: Create a subtask under this task
• success(): Mark task as successful
• failure(error: Optional[str] = None): Mark task as failed
• set_message(message: Optional[str]): Set a message for the task
• get_from_redis(redis: Redis, *, key: str) -> Optional["ProgressTask"]: Class method to retrieve a task from Redis
• get_prebuilt_subtasks_model(root: "ProgressTask", redis: Redis, *, model_key: str) -> List["ProgressTask"]: Class method to get prebuilt subtasks

Schemas

ProgressTaskStage

Enum with the following values:

• PENDING: Task is in progress
• SUCCESS: Task completed successfully
• FAILURE: Task failed

ProgressReport

• progress_key: Optional[str]: Key used to store progress in Redis
• progress: float: Completion percentage (0.0 to 1.0)
• is_root_success: bool: Whether the root task is successful
• success: int: Number of successful tasks
• failure: int: Number of failed tasks
• pending: int: Number of pending tasks
• assigned: int: Number of assigned tasks
• total: int: Total number of tasks
• message: Optional[str]: Optional message
• errors: List[str]: List of error messages

Usage Examples

import uuid
from redis import Redis
from keble_helpers import ProgressHandler

# Initialize Redis connection
redis = Redis(host='localhost', port=6379, db=0)

# Create a progress handler
handler = ProgressHandler(redis=redis)

# Create a new progress task
task_key = str(uuid.uuid4())
task = handler.new(key=task_key)

# Create subtasks
subtask1 = task.new_subtask()
subtask2 = task.new_subtask()
subtask3 = task.new_subtask()

# Mark tasks as complete or failed
subtask1.success()
subtask2.failure(error="Something went wrong")
subtask3.success()
task.success()

# Get progress report
report = handler.get(key=task_key)
print(f"Progress: {report.progress * 100}%")
print(f"Success: {report.success}, Failure: {report.failure}, Pending: {report.pending}")

# Using model_key for prebuilt subtasks
model_key = str(uuid.uuid4())
root_task = handler.new(key=str(uuid.uuid4()), model_key=model_key)

# When you create a new task with the same model_key,
# it will have the same number of subtasks
new_task = handler.new(key=str(uuid.uuid4()), model_key=model_key)

Pydantic

The Pydantic module provides helpers and utilities for working with Pydantic models.

Functions

• is_http_url(url: Any) -> bool: Validates if a string is a valid HTTP or HTTPS URL by checking if it has a valid HTTP/HTTPS scheme and netloc

Base Classes

PydanticModelConfig

• default_dict(**kwargs) -> dict: Returns a dictionary with default configuration
• default(**kwargs) -> ConfigDict: Returns a ConfigDict with default configuration

CloudStorageBase

• Base model for cloud storage objects with standardized fields

Enums

CloudStorageType

• AWS_S3: Amazon S3 storage
• ALIYUN_OSS: Alibaba Cloud OSS storage

CloudStorageObjectType

• IMAGE: Image files
• VIDEO: Video files
• EXCEL: Excel spreadsheets
• CSV: CSV files
• OTHER: Other file types
• determine_type(*, mime: str) -> CloudStorageObjectType: Determine type from MIME

Usage Examples

from keble_helpers.pydantic import CloudStorageBase, CloudStorageObjectType, CloudStorageType
from keble_helpers.pydantic.schemas import is_http_url
from pydantic import BaseModel

# Check if a URL is valid HTTP/HTTPS
valid = is_http_url("https://example.com")  # True
valid = is_http_url("ftp://example.com")    # False
valid = is_http_url("example.com")          # False (missing scheme)
valid = is_http_url("http://")              # False (missing netloc)

# Create a custom model with Pydantic configuration
class MyModel(BaseModel):
    model_config = PydanticModelConfig.default()
    # Fields go here

# Create a cloud storage object
storage = CloudStorageBase(
    key="path/to/file.jpg",
    base_url="https://example.com/storage",
    type=CloudStorageType.AWS_S3,
    object_type=CloudStorageObjectType.IMAGE,
    original_file_name="photo.jpg"
)

# Determine object type from MIME
object_type = CloudStorageObjectType.determine_type(mime="image/jpeg")

Common

The Common module provides general utility functions for common tasks.

Functions

String and ID Utilities

• id_generator() -> str: Generate a UUID4 string
• generate_random_string(length: int = 32, *, lower: bool = True, upper: bool = True, digit: bool = True) -> str: Generate a random string
• hash_string(arg: str) -> str: Generate MD5 hash of a string
• inline_string(string: str, max_len: int = 30): Format a string for inline display

Pydantic Helpers

• is_pydantic_field_empty(obj: BaseModel, field: str) -> bool: Check if a field is empty in a Pydantic model

Date and Time

• date_to_datetime(d: date) -> datetime: Convert a date to datetime
• datetime_to_date(d: datetime) -> date: Convert a datetime to date

List and Collection Operations

• slice_to_list(items: List[Any], slice_size: int) -> List[List[Any]]: Split a list into chunks
• get_first_match(items: list, key_fn, value): Find first item in a list matching a criterion

File System Operations

• ensure_has_folder(path: str) -> str: Create a directory if it doesn't exist
• yield_files(folder: str) -> Iterator[str | Path]: Recursively yield files in a directory
• get_files(folder: str) -> List[str | Path]: Get a list of all files in a directory
• zip_dir(folder: Path | str, zip_filepath: Path | str): Zip a directory
• remove_dir(dir: Path | str): Remove a directory

MIME Type Checking

• is_mime_prefix_in(mime, mime_start: List[str]): Check if a MIME type has a specific prefix
• is_mime_image(mime: str): Check if a MIME type is an image
• is_mime_video(mime: str): Check if a MIME type is a video
• is_mime_audio(mime: str): Check if a MIME type is audio
• is_mime_media(mime: str): Check if a MIME type is any media (image, video, audio)
• is_mime_ms_excel(mime: str): Check if a MIME type is MS Excel
• is_mime_csv(mime: str): Check if a MIME type is CSV

Usage Examples

from keble_helpers.common import (
    id_generator, hash_string, ensure_has_folder, get_files, 
    is_mime_image, slice_to_list
)

# Generate a unique ID
unique_id = id_generator()

# Generate a hash of a string
file_hash = hash_string("content to hash")

# Ensure a directory exists
path = ensure_has_folder("/path/to/directory")

# Get all files in a directory
files = get_files("/path/to/directory")

# Check if a MIME type is an image
is_image = is_mime_image("image/jpeg")  # True

# Split a list into chunks of size 3
chunks = slice_to_list([1, 2, 3, 4, 5, 6, 7], 3)  # [[1, 2, 3], [4, 5, 6], [7]]

DateTime

The DateTime module provides utilities for working with dates and times.

Functions

• days_in_month(year, month): Get the number of days in a specific month

Usage Examples

from keble_helpers.datetime import days_in_month

# Get days in February 2024 (leap year)
days = days_in_month(2024, 2)  # 29

# Get days in February 2023 (non-leap year)
days = days_in_month(2023, 2)  # 28

Enum

The Enum module provides predefined enumerations.

Enums

Environment

• development: Development environment
• test: Test environment
• production: Production environment

Usage Examples

from keble_helpers.enum import Environment

# Use environment enum
current_env = Environment.development

# Check environment
if current_env == Environment.production:
    # Production-specific code
    pass

FastAPI

The FastAPI module provides helpers for working with FastAPI applications, focusing on JSON encoding compatible with Pydantic v2.

Functions

• jsonable_encoder(obj: Any, include: Optional[IncEx] = None, exclude: Optional[IncEx] = None, by_alias: bool = True, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, custom_encoder: Optional[Dict[Any, Callable[[Any], Any]]] = None, sqlalchemy_safe: bool = True) -> Any: Convert a Python object to a JSON-compatible object

Constants

• PYDANTIC_V2: Boolean indicating if Pydantic v2 is in use
• ENCODERS_BY_TYPE: Dictionary mapping Python types to encoder functions

Usage Examples

from keble_helpers.fastapi import jsonable_encoder
from pydantic import BaseModel
from datetime import datetime

class User(BaseModel):
    id: int
    name: str
    created_at: datetime
    updated_at: datetime | None = None

user = User(id=1, name="John Doe", created_at=datetime.now())

# Convert to JSON-compatible dict
json_data = jsonable_encoder(user)

# Convert excluding some fields
json_data = jsonable_encoder(user, exclude={"created_at"})

# Convert with custom encoders
json_data = jsonable_encoder(
    user, 
    custom_encoder={datetime: lambda dt: dt.strftime("%Y-%m-%d")}
)

File

The File module provides utilities for file operations, particularly for downloading files.

Functions

• adownload_file(*, url: str, folder: Path, filename: str) -> Path: Asynchronously download a file from a URL

Usage Examples

import asyncio
from pathlib import Path
from keble_helpers.file import adownload_file

async def download_example():
    # Download a file
    file_path = await adownload_file(
        url="https://example.com/file.pdf",
        folder=Path("/path/to/downloads"),
        filename="document.pdf"
    )
    
    print(f"Downloaded to: {file_path}")

# Run the async function
asyncio.run(download_example())

Multithread (Deprecated)

Note: This module is deprecated. The project now uses async-based approaches instead of multithreading.

The Multithread module provides utilities for thread management and parallel execution.

Classes

ThreadController

• __init__(thread_size: int): Initialize with a maximum number of threads
• create_thread(target: Callable, *, args: Optional[tuple] = None, kwargs: Optional[Dict[str, Any]] = None, thread_owner: Optional[str | int] = None, disable_sema: Optional[bool] = False, join: Optional[bool] = False): Create and start a new thread
• acquire(*, thread_owner: Optional[str | int] = None): Acquire a semaphore
• release(*, thread_owner: Optional[str | int] = None): Release a semaphore
• wait_all_to_finish(): Wait for all threads to complete
• wait_owner_to_finish(thread_owner: str | int): Wait for all threads by a specific owner to complete

Decorators

• threaded(*, sema: Optional[Semaphore] = None, join: Optional[bool] = False): Decorator to run a function in a separate thread

Usage Examples

from keble_helpers import ThreadController, threaded
from threading import Semaphore

# Using ThreadController
controller = ThreadController(thread_size=5)

def task(results):
    # Perform task
    results.append("Task completed")
    controller.release()

results = []
for _ in range(10):
    controller.create_thread(target=task, args=(results,))

controller.wait_all_to_finish()

# Using threaded decorator
sema = Semaphore(3)

@threaded(sema=sema)
def background_task(results):
    results.append("Background task completed")
    sema.release()

threads = []
results = []
for _ in range(5):
    threads.append(background_task(results))

for thread in threads:
    thread.join()

NumPy Utils

The NumPy Utils module provides helper functions for working with NumPy arrays and handling numerical values.

Functions

• is_invalid_float(value: Optional[float]) -> bool: Check if a float value is NaN or infinity
• guard_invalid_float(value: float | None | np.floating) -> float | None: Replace invalid float values (NaN, inf) with None

Usage Examples

import numpy as np
from keble_helpers.np_utils import is_invalid_float, guard_invalid_float

# Check if a value is an invalid float
invalid = is_invalid_float(float('nan'))  # True
invalid = is_invalid_float(float('inf'))  # True
invalid = is_invalid_float(42.0)  # False

# Guard against invalid floats
safe_value = guard_invalid_float(np.nan)  # None
safe_value = guard_invalid_float(np.inf)  # None
safe_value = guard_invalid_float(42.0)  # 42.0
safe_value = guard_invalid_float(np.float32(3.14))  # 3.14

Pydantic AI image runtime

keble_helpers.ai.image_runtime provides reusable multimodal runtime hardening helpers for any pydantic-ai callsite that may include typed ImageUrl parts.

Key exports:

• ImagePromptChecker: bounded-concurrency URL preflight with TTL cache and provider-safe image-count budgeting
• ImagePreflightDecision, ImagePreflightBatchResult, ImagePreflightReason: typed preflight decisions
• extract_image_urls(...), strip_image_url_parts(...)
• is_image_url_model_404_error(...)
• arun_with_image_fallback(...): preflight + image-404 text-only fallback retry (tenacity with reraise=True)
• typed_tool(...), typed_tool_plain(...): additive wrappers around agent.tool(...) / agent.tool_plain(...) that preserve the decorated callable type for downstream code

Ownership model:

• host applications should instantiate one shared ImagePromptChecker
• downstream libs should accept that instance and reuse it
• retry policy and preflight policy live on the same checker instance
• libs should not construct hidden per-module checker objects with divergent settings

Status policy (ImagePromptChecker):

• only_accepts: list[int] | None
• rejects: list[int] | None
• provide only one of them (ValueError if both are provided)
• when both are omitted, default behavior is only_accepts=[200, 206]
• probe requests do not auto-follow redirects; 3xx status codes are surfaced to policy evaluation directly

Usage:

from keble_helpers import ImagePromptChecker, arun_with_image_fallback

checker = ImagePromptChecker(
    enabled=True,
    timeout_secs=2.0,
    max_concurrency=4,
    cache_ttl_secs=600,
    max_images_per_prompt=45,
    image_model_404_retry_attempts=1,
    only_accepts=[200, 206],  # strict image status policy
)

result = await arun_with_image_fallback(
    agent=agent,
    prompt=prompt_parts,  # Sequence[UserContent] with optional ImageUrl
    image_prompt_checker=checker,
)
output = result.output  # preserves the agent's concrete output type

# or call through the checker directly
result = await checker.arun_with_image_fallback(
    agent=agent,
    prompt=prompt_parts,
)

Typed tool registration:

from pydantic_ai import Agent
from pydantic_ai.tools import RunContext
from keble_helpers import typed_tool, typed_tool_plain

agent = Agent("test", deps_type=int, output_type=str)

@typed_tool(agent, require_parameter_descriptions=True)
async def repeat(ctx: RunContext[int], word: str) -> str:
    """Repeat a word.

    Args:
        word: Word to repeat.
    """

    return f"{ctx.deps}:{word}"


@typed_tool_plain(agent, name="slugify")
def slugify(name: str) -> str:
    return name.strip().lower().replace(" ", "-")

Contract:

1. Keep domain models/validators pure (no network I/O in validators).
2. Apply runtime preflight right before model invocation.
3. Keep retry ownership on the injected checker instead of separate function kwargs.
4. Preserve terminal exception surfaces (reraise=True) for upstream task/error layers.

Usage Accounting Events

keble_helpers.ai.usage_accounting defines package-neutral usage facts only. Packages should build events with UsageAccountingEvent.from_run_usage(...), from_request_usage(...), or from_unit_count(...) and pass them to an injected UsageAccountingRecorderProtocol.

This helper layer must not import Mongo, keble-task, pricing catalogs, or backend business logic. Sources and units are uppercase Enums so downstream packages and backend recorders share one typed contract.

Use UsageAccountingEvent.from_result_usage(...) when a package records a Pydantic-AI agent result. The classmethod owns normalization for both result.usage value-style APIs and result.usage() method-style APIs, keeping runtime compatibility out of downstream business logic.