Monitoring & Performance

import logging
import time
import pixeltable as pxt

logger = logging.getLogger(__name__)

@pxt.udf
def process_video(video: pxt.Video) -> pxt.Json:
    start = time.time()
    try:
        # Your processing logic here
        result = {'processed': True}
        logger.info(f"Processed in {time.time() - start:.2f}s")
        return result
    except Exception as e:
        logger.error(f"Processing failed: {e}")
        raise

# Batch UDF execution for GPU models
@pxt.udf(batch_size=32)
def embed_batch(texts: pxt.Batch[str]) -> pxt.Batch[list[float]]:
    # Process multiple items at once
    return model.encode(texts)

# Batch inserts (more efficient than individual inserts)
table.insert([row1, row2, row3, ...])

# ~/.pixeltable/config.toml
[openai]
requests_per_minute = 500
tokens_per_minute = 90000

# Default: 600 requests per minute
@pxt.udf(resource_pool='request-rate:my_service')
async def call_custom_api(prompt: str) -> dict:
    # Your logic to call custom endpoint
    return await custom_api_call(prompt)

# Example: Custom rate-limited UDF for self-hosted model
@pxt.udf(resource_pool='request-rate:my_ray_cluster')
async def call_ray_model(prompt: str, model: str) -> dict:
    # Your logic to call FastAPI + Ray cluster
    return await custom_api_call(prompt, model)