ApiLib Client Documentation
Table of Contents
- Overview
- Installation
- Quick Start
- Configuration
- Basic CRUD Operations
- Field-Level Encryption
- Advanced Features
- Error Handling
- API Reference
- Examples
Overview
ApiLib is a Python client library for interacting with the domain-based multi-tenant API server. It provides:
- Automatic object serialization/deserialization using pymodel schema classes
- Dynamic method generation for all object types
- Optional field-level encryption for sensitive data
- Multi-tenant support with domain isolation
- Extension system for custom functionality
- Factory methods for easy configuration
- Singleton pattern support for shared instances
Architecture
┌─────────────────────────────────────┐
│ Your Application │
│ │
│ from py_api_lib import ApiLib │
│ api = ApiLib(...) │
│ backend = api.mcp_backend_read() │
└─────────────────────────────────────┘
↓
┌─────────────────────────────────────┐
│ ApiLib (py_api_lib) │
│ - Serialization │
│ - Encryption (optional) │
│ - Dynamic methods │
│ - Error handling │
└─────────────────────────────────────┘
↓ HTTP/REST
┌─────────────────────────────────────┐
│ API Server │
│ - Schema validation │
│ - Multi-tenant isolation │
│ - MongoDB storage │
└─────────────────────────────────────┘
Installation
Prerequisites
# Python 3.8 or higher
python --version
# Install dependencies
pip install requests inflection gevent
Install from Wheel
# Install py_api_lib
pip install py_api_lib-1.0.0-py3-none-any.whl
# Install tenant schema (required)
pip install py_api_e_commerce-0.1.0-py3-none-any.whl
Install for Development
cd infra/libs/py
pip install -e .
Optional: Encryption Support
# Required only if using encryption
pip install cryptography==41.0.0
Quick Start
Basic Usage (No Encryption)
from py_api_lib import ApiLib
from pymodel.objects.mcp_backend import McpBackend
# Create ApiLib instance
api = ApiLib(
api_server_host='localhost',
api_server_port='8082'
)
# Create an object
backend = McpBackend(
name='my-backend',
fq_name=['my-domain', 'my-project', 'my-backend'],
parent_uuid='parent-uuid-here',
api_endpoint='https://api.example.com',
description='My backend service'
)
# Save to server
uuid = api.mcp_backend_create(backend)
print(f"Created: {uuid}")
# Read back
retrieved = api.mcp_backend_read(id=uuid)
print(f"Retrieved: {retrieved.name}")
# Update
retrieved.set_description('Updated description')
api.mcp_backend_update(retrieved)
# List all
backends = api.mcp_backends_list(detail=True)
print(f"Total backends: {len(backends)}")
# Delete
api.mcp_backend_delete(id=uuid)
With Encryption
from py_api_lib import ApiLib
from py_api_lib.encryption import generate_key
import base64
# Generate encryption key
key_b64 = generate_key()
key = base64.b64decode(key_b64)
# Create ApiLib with encryption enabled
api = ApiLib(
api_server_host='localhost',
api_server_port='8082',
enable_encryption=True,
encryption_key=key,
domain_name='my-domain'
)
# Use exactly as before - encryption is transparent!
backend = McpBackend(
name='secure-backend',
fq_name=['my-domain', 'my-project', 'secure-backend'],
parent_uuid='parent-uuid',
api_key='secret-key-12345', # Will be ENCRYPTED
config={'db': 'postgres'} # Will be ENCRYPTED
)
uuid = api.mcp_backend_create(backend)
retrieved = api.mcp_backend_read(id=uuid) # Automatically DECRYPTED
Configuration
Configuration Methods
1. Direct Instantiation
api = ApiLib(
api_server_host='localhost',
api_server_port='8082',
timeout=30,
max_retries=3,
enable_encryption=False
)
2. From Configuration File
# config.yaml
api_lib:
api_server_host: localhost
api_server_port: 8082
timeout: 30
max_retries: 3
encryption:
enabled: true
key: "SGVsbG9Xb3JsZDEyMzQ1Njc4OTBBQkNERUZHSElKS0w=" # base64
domain: "my-domain"
api = ApiLib.from_config('config.yaml')
3. From Environment Variables
export API_SERVER_HOST=localhost
export API_SERVER_PORT=8082
export ENCRYPTION_ENABLED=true
export ENCRYPTION_KEY="SGVsbG9Xb3JsZDEyMzQ1Njc4OTBBQkNERUZHSElKS0w="
export DOMAIN_NAME=my-domain
api = ApiLib.from_config() # Loads from environment
4. Singleton Pattern
from py_api_lib import ApiLibSingleton
# Initialize once (e.g., at application startup)
ApiLibSingleton.initialize(
api_server_host='localhost',
api_server_port='8082'
)
# Use anywhere in your application
from py_api_lib import get_api_lib
api = get_api_lib()
Configuration Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
api_server_host |
str | '127.0.0.1' |
API server hostname/IP |
api_server_port |
str | '8082' |
API server port |
timeout |
int | 30 |
Request timeout (seconds) |
max_retries |
int | 3 |
Maximum retry attempts |
enable_encryption |
bool | False |
Enable field-level encryption |
encryption_key |
bytes | None |
32-byte AES-256 key (required if encryption enabled) |
domain_name |
str | None |
Domain/tenant name (for logging) |
wait_for_connect |
bool | False |
Wait for server connection on init |
extensions |
list | None |
List of extensions to load |
Basic CRUD Operations
ApiLib dynamically creates methods for all object types in your schema.
Method Naming Convention
For an object type mcp-backend (or mcp_backend):
- Create:
mcp_backend_create(obj) - Read:
mcp_backend_read(id=None, fq_name=None) - Update:
mcp_backend_update(obj) - Delete:
mcp_backend_delete(id=None, fq_name=None) - List:
mcp_backends_list(detail=False, parent_id=None, ...)
Create
from pymodel.objects.project import Project
project = Project(
name='my-project',
fq_name=['my-domain', 'my-project'],
parent_uuid='domain-uuid',
description='My project'
)
uuid = api.project_create(project)
Returns: UUID string of created object
Read
# Read by UUID
project = api.project_read(id='uuid-here')
# Read by fq_name (list)
project = api.project_read(fq_name=['my-domain', 'my-project'])
# Read by fq_name (string)
project = api.project_read(fq_name_str='my-domain:my-project')
Returns: Object instance or None if not found
Update
# Read, modify, update
project = api.project_read(id='uuid-here')
project.set_description('Updated description')
api.project_update(project)
Returns: "Updated successfully"
Delete
# Delete by UUID
api.project_delete(id='uuid-here')
# Delete by fq_name
api.project_delete(fq_name=['my-domain', 'my-project'])
Returns: None
List
# List all (returns raw data)
projects = api.projects_list()
# List all with detail (returns object instances)
projects = api.projects_list(detail=True)
# List by parent
projects = api.projects_list(
detail=True,
parent_id='domain-uuid'
)
# List specific objects
projects = api.projects_list(
detail=True,
obj_uuids=['uuid1', 'uuid2']
)
Returns: List of objects (detail=True) or raw data dicts (detail=False)
Bulk List
# List multiple objects efficiently
backends = api.list_bulk(
'mcp_backend',
uuids=['uuid1', 'uuid2', 'uuid3']
)
# List all with filters
backends = api.list_bulk(
'mcp_backend',
filters='status==active'
)
Field-Level Encryption
Overview
Field-level encryption protects sensitive data while allowing the server to:
- Validate object structure
- Query by metadata (uuid, fq_name, name)
- Enforce multi-tenant isolation
- Maintain referential integrity
What Gets Encrypted?
Never Encrypted (Required for Server):
uuid,fq_name,parent_uuidobj_type,parent_typename(kept searchable)topic
Always Encrypted (When Enabled):
- All other schema-defined fields
- API keys, passwords, tokens
- Configuration values
- Any business-sensitive data
Enable Encryption
from py_api_lib.encryption import generate_key
import base64
# Generate a new key
key_b64 = generate_key()
print(f"Store this securely: {key_b64}")
# Use the key
key = base64.b64decode(key_b64)
api = ApiLib(
enable_encryption=True,
encryption_key=key,
domain_name='my-domain'
)
Key Management
from py_api_lib.encryption import EncryptionManager
# Generate key
key = EncryptionManager._generate_key() # 32 bytes
# Derive key from password
password = "my-secure-password"
key, salt = EncryptionManager.derive_key_from_password(password)
# Store salt securely for key recovery
Multi-Tenant Encryption
from py_api_lib.encryption import EncryptionConfig
# Load config with multiple domain keys
config = EncryptionConfig.from_env()
# Create separate ApiLib per domain
for domain in ['domain1', 'domain2', 'domain3']:
key = config.get_key_for_domain(domain)
api_instances[domain] = ApiLib(
enable_encryption=True,
encryption_key=key,
domain_name=domain
)
Encryption Error Handling
from py_api_lib import EncryptionRequiredError
try:
# Client without encryption tries to read encrypted object
api_plain = ApiLib(enable_encryption=False)
obj = api_plain.mcp_backend_read(id='encrypted-object-uuid')
except EncryptionRequiredError as e:
print(f"Error: {e}")
# Output: Object contains encrypted data.
# To read this object, create ApiLib with:
# enable_encryption=True
# encryption_key=<your-32-byte-key>
Advanced Features
Reference Management
# Add reference
api.ref_update(
obj_type='mcp_backend',
obj_uuid='backend-uuid',
ref_type='mcp_tool',
ref_uuid='tool-uuid',
ref_fq_name=['domain', 'project', 'tool'],
operation='ADD'
)
# Delete reference
api.ref_update(
obj_type='mcp_backend',
obj_uuid='backend-uuid',
ref_type='mcp_tool',
ref_uuid='tool-uuid',
ref_fq_name=['domain', 'project', 'tool'],
operation='DELETE'
)
Query Operations
import json
query_data = {
"op": "find-sub-classes",
"lon": -122.4194,
"lat": 37.7749
}
results = api.query(json.dumps(query_data))
FQ Name ↔ UUID Conversion
# FQ name to UUID
uuid = api.fq_name_to_uuid('project', ['domain', 'project'])
# UUID to object
obj_uuid = api.obj_to_id(project_object)
Object Utilities
# Convert object to JSON
json_str = api.obj_to_json(project)
# Convert object to dict
obj_dict = api.obj_to_dict(project)
# Get object class
ProjectClass = api.get_obj_class('project')
# Create object instance
project = api.create_object_instance(
'project',
name='my-project',
parent=domain_object,
description='My description'
)
Extensions
# Load extension
api.load_extension('my_custom_extension')
# List loaded extensions
extensions = api.get_loaded_extensions()
# Check if extension loaded
if api.has_extension('my_extension'):
# Extension-specific functionality
pass
Context Manager
with ApiLib(api_server_host='localhost') as api:
backend = api.mcp_backend_read(id='uuid')
# Automatically closes connection on exit
Error Handling
Exception Hierarchy
from py_api_lib.exceptions import (
ApiException, # Base exception
NotFoundException, # 404 errors
ConflictException, # 409 errors
ValidationException, # Validation errors
BadRequestException # 400 errors
)
from py_api_lib import EncryptionRequiredError
from pymodel.serialization_errors import (
SerializationError,
ObjectSerializationError,
ObjectDeserializationError
)
Error Handling Examples
from py_api_lib.exceptions import NotFoundException, ValidationException
from py_api_lib import EncryptionRequiredError
try:
backend = api.mcp_backend_read(id='non-existent-uuid')
except NotFoundException as e:
print(f"Object not found: {e}")
except EncryptionRequiredError as e:
print(f"Encryption required: {e}")
print("Encrypted fields:", e.args[0])
except ValidationException as e:
print(f"Validation failed: {e}")
except ApiException as e:
print(f"API error: {e}")
except Exception as e:
print(f"Unexpected error: {e}")
API Reference
ApiLib Class
Constructor
ApiLib(
api_server_host='127.0.0.1',
api_server_port='8082',
timeout=30,
max_retries=3,
enable_encryption=False,
encryption_key=None,
domain_name=None,
wait_for_connect=False,
extensions=None
)
Factory Methods
# From configuration file
api = ApiLib.from_config('config.yaml', **overrides)
# Get default instance (singleton)
api = ApiLib.get_default_instance()
# Reset default instance
ApiLib.reset_default_instance()
Dynamic Methods (Generated for Each Object Type)
For object type {obj_type}:
# Create
uuid = api.{obj_type}_create(obj)
# Read
obj = api.{obj_type}_read(id=None, fq_name=None, fq_name_str=None)
# Update
result = api.{obj_type}_update(obj)
# Delete
api.{obj_type}_delete(id=None, fq_name=None)
# List
objects = api.{obj_type}s_list(
detail=False,
parent_id=None,
parent_fq_name=None,
obj_uuids=None,
filters=None
)
# Get default ID
default_id = api.{obj_type}_get_default_id()
Utility Methods
# FQ name conversion
uuid = api.fq_name_to_uuid(obj_type, fq_name)
# Object conversion
json_str = api.obj_to_json(obj)
obj_dict = api.obj_to_dict(obj)
uuid = api.obj_to_id(obj)
# Reference management
result = api.ref_update(obj_type, obj_uuid, ref_type, ref_uuid, ref_fq_name, operation)
# Query
results = api.query(json_data)
# Bulk operations
objects = api.list_bulk(obj_type, uuids=None, **kwargs)
# Health check
is_healthy = api.health_check()
# Get object class
ObjClass = api.get_obj_class(obj_type)
# Create instance
obj = api.create_object_instance(obj_type, name, parent=None, **kwargs)
Singleton Manager
from py_api_lib import ApiLibSingleton, get_api_lib, initialize_api_lib
# Initialize
ApiLibSingleton.initialize(**config)
# Or use convenience function
initialize_api_lib(**config)
# Get instance
api = ApiLibSingleton.get_instance()
# Or use convenience function
api = get_api_lib()
# Check if initialized
if ApiLibSingleton.is_initialized():
api = ApiLibSingleton.get_instance()
# Health check
is_healthy = ApiLibSingleton.health_check()
# Reset
ApiLibSingleton.reset()
Examples
Example 1: Complete CRUD Workflow
from py_api_lib import ApiLib
from pymodel.objects.domain import Domain
from pymodel.objects.project import Project
from pymodel.objects.mcp_backend import McpBackend
# Initialize
api = ApiLib(api_server_host='localhost', api_server_port='8082')
# 1. Create domain
domain = Domain(
name='my-domain',
fq_name=['my-domain'],
parent_uuid='config-root-uuid'
)
domain_uuid = api.domain_create(domain)
print(f"Created domain: {domain_uuid}")
# 2. Create project
project = Project(
name='my-project',
fq_name=['my-domain', 'my-project'],
parent_uuid=domain_uuid,
description='My project'
)
project_uuid = api.project_create(project)
print(f"Created project: {project_uuid}")
# 3. Create backend
backend = McpBackend(
name='my-backend',
fq_name=['my-domain', 'my-project', 'my-backend'],
parent_uuid=project_uuid,
api_endpoint='https://api.example.com',
description='My backend'
)
backend_uuid = api.mcp_backend_create(backend)
print(f"Created backend: {backend_uuid}")
# 4. Read and update
backend = api.mcp_backend_read(id=backend_uuid)
backend.set_description('Updated description')
api.mcp_backend_update(backend)
print("Backend updated")
# 5. List all backends in project
backends = api.mcp_backends_list(detail=True, parent_id=project_uuid)
print(f"Found {len(backends)} backends")
# 6. Clean up
api.mcp_backend_delete(id=backend_uuid)
api.project_delete(id=project_uuid)
api.domain_delete(id=domain_uuid)
print("Cleanup complete")
Example 2: Encryption with Multiple Domains
from py_api_lib import ApiLib
from py_api_lib.encryption import generate_key
import base64
# Generate separate keys for each domain
domain_keys = {
'domain1': base64.b64decode(generate_key()),
'domain2': base64.b64decode(generate_key()),
'domain3': base64.b64decode(generate_key())
}
# Create ApiLib instances per domain
api_instances = {}
for domain, key in domain_keys.items():
api_instances[domain] = ApiLib(
api_server_host='localhost',
api_server_port='8082',
enable_encryption=True,
encryption_key=key,
domain_name=domain
)
# Use domain-specific clients
api_domain1 = api_instances['domain1']
backend1 = api_domain1.mcp_backend_create(...) # Encrypted with domain1 key
api_domain2 = api_instances['domain2']
backend2 = api_domain2.mcp_backend_create(...) # Encrypted with domain2 key
# Domain isolation: domain1 can't decrypt domain2's data
Example 3: Error Recovery
from py_api_lib import ApiLib, EncryptionRequiredError
from py_api_lib.exceptions import NotFoundException
api_encrypted = ApiLib(
enable_encryption=True,
encryption_key=my_key
)
api_plain = ApiLib(enable_encryption=False)
def safe_read(api, obj_type, obj_id):
"""Read object with fallback to encrypted client."""
try:
return api.mcp_backend_read(id=obj_id)
except EncryptionRequiredError:
print("Object is encrypted, retrying with encryption...")
return api_encrypted.mcp_backend_read(id=obj_id)
except NotFoundException:
print("Object not found")
return None
# Use it
backend = safe_read(api_plain, 'mcp_backend', 'uuid-here')
Example 4: Application Initialization
# app_init.py
import os
import base64
from py_api_lib import initialize_api_lib
def init_api_client():
"""Initialize API client at application startup."""
# Load configuration from environment
config = {
'api_server_host': os.getenv('API_SERVER_HOST', 'localhost'),
'api_server_port': os.getenv('API_SERVER_PORT', '8082'),
'timeout': int(os.getenv('API_TIMEOUT', '30')),
}
# Add encryption if enabled
if os.getenv('ENCRYPTION_ENABLED', '').lower() == 'true':
encryption_key = os.getenv('ENCRYPTION_KEY')
if not encryption_key:
raise ValueError("ENCRYPTION_KEY required when encryption enabled")
config['enable_encryption'] = True
config['encryption_key'] = base64.b64decode(encryption_key)
config['domain_name'] = os.getenv('DOMAIN_NAME', 'default')
# Initialize singleton
return initialize_api_lib(**config)
# Initialize once at startup
api = init_api_client()
# anywhere in application
from py_api_lib import get_api_lib
def create_backend(name, endpoint):
api = get_api_lib() # Get shared instance
backend = McpBackend(...)
return api.mcp_backend_create(backend)
Best Practices
1. Use Encryption for Sensitive Data
# DO: Use encryption for production data
api = ApiLib(
enable_encryption=True,
encryption_key=securely_loaded_key
)
2. Store Keys Securely
# DON'T: Hardcode keys
encryption_key = base64.b64decode("SGVsbG9...") # BAD!
# DO: Load from secure storage
from your_vault import get_encryption_key
encryption_key = get_encryption_key('my-domain')
3. Use Singleton for Application-Wide Client
# DO: Initialize once, use everywhere
initialize_api_lib(...)
# In different modules
api = get_api_lib()
4. Handle Encryption Errors Gracefully
# DO: Provide helpful feedback
try:
obj = api.read(id=uuid)
except EncryptionRequiredError as e:
logger.error(f"Cannot read encrypted object: {e}")
print("Please enable encryption with the correct key")
5. Use Context Managers for Cleanup
# DO: Ensure cleanup
with ApiLib(...) as api:
backend = api.mcp_backend_create(...)
# Connection automatically closed
Troubleshooting
Encryption Errors
# Error: Domain key must be exactly 32 bytes
# Solution: Ensure key is 32 bytes for AES-256
key = base64.b64decode(key_b64)
assert len(key) == 32
Connection Errors
# Error: Connection refused
# Solution: Check server is running
api = ApiLib(wait_for_connect=True, timeout=60)
Support
For issues or questions:
- Check server logs for detailed error messages
- Review API server documentation
- Contact support team
Version History
- 1.0.0: Initial release with encryption support
- Field-level value encryption
- Multi-tenant key isolation
- Backward compatible with existing code
Release files for py-api-lib-system 1.0.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| py_api_lib_system-1.0.0-py3-none-any.whl | Python 3 | none | any | Details |
Release files / py_api_lib_system-1.0.0-py3-none-any.whl
| Download URL | py_api_lib_system-1.0.0-py3-none-any.whl |
|---|---|
| Size | 62.0 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
2a4031ab3461547895dc0d74ae4f6a34796a828d90bc64677354e3ce544feb82
|
|
BLAKE2b-256 checksum How to use checksums |
d91fb380d93266ea23ad42a8ed5f80689f2941ad1eea1c7a34a549b939b136b5
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/6.2.0 CPython/3.10.12
|