apifrom

APIFromAnything - Transform any Python function into a REST API endpoint.

This library allows developers to expose ordinary Python functions as API endpoints with minimal boilerplate, automatically handling routing, input validation, output serialization, error handling, and documentation generation.

Overview

Classes

• API

• APIProfiler

• AdvancedCacheMiddleware

• BatchProcessor

• CacheAnalytics

• CacheBackend

• CacheOptimizer

• ConnectionPool

• ConnectionPoolMiddleware

• HybridEvictionPolicy

• JSONExporter

• LFUEvictionPolicy

• LRUEvictionPolicy

• LogExporter

• MemoryCacheBackend

• MetricsCollector

• MetricsMiddleware

• Middleware

• OptimizationAnalyzer

• OptimizationConfig

• OptimizationLevel

• OptimizationRecommendation

• OptimizedCacheMiddleware

• PoolManager

• ProfileMiddleware

• ProfileReport

• PrometheusExporter

• RedisCacheBackend

• RequestCoalescingMiddleware

• SizeEvictionPolicy

• TTLEvictionPolicy

• Web

Functions

• api()

• api_key_required()

• basic_auth_required()

• batch_process()

• coalesce_requests()

• jwt_required()

• oauth2_required()

Classes

class apifrom.API(debug=False, title='APIFromAnything API', description='API created with APIFromAnything', version='1.0.0', docs_url='/docs', openapi_config=None, swagger_ui_config=None, enable_docs=True)

Parameters:

• debug (bool)

• title (str)

• description (str)

• version (str)

• docs_url (str)

• openapi_config (apifrom.docs.openapi.OpenAPIConfig | None)

• swagger_ui_config (apifrom.docs.swagger_ui.SwaggerUIConfig | None)

• enable_docs (bool)

Main application container for APIFromAnything.

This class serves as the central registry for all API endpoints, middleware, and configuration. It provides methods for registering endpoints, middleware, and starting the server.

apifrom.router

The router instance for managing routes.

Type:

Router

apifrom.middleware

List of middleware instances.

Type:

list

apifrom.debug

Whether to run in debug mode.

Type:

bool

apifrom.title

The title of the API.

Type:

str

apifrom.description

The description of the API.

Type:

str

apifrom.version

The version of the API.

Type:

str

apifrom.docs_url

The URL for the API documentation.

Type:

str

Initialize a new API instance.

param debug:

Whether to run in debug mode.

param title:

The title of the API.

param description:

The description of the API.

param version:

The version of the API.

param docs_url:

The URL for the API documentation.

param openapi_config:

Configuration for OpenAPI documentation.

param swagger_ui_config:

Configuration for Swagger UI.

param enable_docs:

Whether to enable API documentation.

apifrom.__call__(scope, receive, send)

ASGI callable.

This method allows the API instance to be used as an ASGI application.

param scope:

The ASGI scope.

param receive:

The ASGI receive function.

param send:

The ASGI send function.

apifrom._build_app()

Build the Starlette application.

returns:

The Starlette application instance.

apifrom._get_current_instance()

:classmethod:

Get the current API instance.

returns:

The current API instance, or None if no instance has been created.

apifrom.add_middleware(middleware)

Add middleware to the API.

param middleware:

The middleware instance to add.

apifrom.add_route(path, endpoint=None, methods=None, name=None, include_in_schema=True, **kwargs)

Add a route to the API.

This method can be used as a decorator or called directly.

param path:

The path for the route.

param endpoint:

The endpoint function.

param methods:

The HTTP methods for the route.

param name:

The name for the route.

param include_in_schema:

Whether to include the route in the OpenAPI schema.

param **kwargs:

Additional arguments to pass to the router.

returns:

The endpoint function if used as a decorator, or None if called directly.

apifrom.add_routes(routes)

Add multiple routes to the API.

This method adds multiple routes to the API at once. The routes should be functions decorated with the @api decorator.

param routes:

A list of functions decorated with the @api decorator.

apifrom.process_request(request)

:async:

Process a request and return a response.

This method is used by adapters to process requests without going through the ASGI interface.

param request:

The request to process.

returns:

The response to the request.

apifrom.register_endpoint(handler, route, method='GET', name=None, **kwargs)

Register an endpoint with the API.

param handler:

The handler function for the endpoint.

param route:

The route for the endpoint.

param method:

The HTTP method for the endpoint.

param name:

The name of the endpoint.

param **kwargs:

Additional arguments to pass to the router.

apifrom.run(host='127.0.0.1', port=8000, **kwargs)

Run the API server.

param host:

The host to bind to.

param port:

The port to bind to.

param **kwargs:

Additional arguments to pass to uvicorn.run.

class apifrom.APIProfiler(output_dir=None, enabled=True)

Parameters:

• output_dir (Optional[str])

• enabled (bool)

Profiles API endpoints to measure performance and identify bottlenecks.

This class provides tools for profiling API endpoints, measuring response times, memory usage, and CPU usage, and generating profile reports.

Initialize an API profiler.

param output_dir:

The directory to save profile reports to (defaults to current directory)

param enabled:

Whether profiling is enabled

apifrom.clear()

Clear all profile data.

apifrom.disable()

Disable profiling.

apifrom.enable()

Enable profiling.

apifrom.get_all_reports()

Get profile reports for all endpoints.

returns:

A list of ProfileReport instances

apifrom.get_report(endpoint_name)

Get a profile report for an endpoint.

param endpoint_name:

The name of the endpoint

returns:

A ProfileReport instance or None if no profile data exists

apifrom.profile_endpoint(endpoint_name)

Decorator for profiling an API endpoint.

param endpoint_name:

The name of the endpoint being profiled

returns:

A decorated function

apifrom.save_reports(prefix=None)

Save all profile reports to files.

param prefix:

A prefix to add to the filenames

returns:

A list of file paths where reports were saved

AdvancedCacheMiddleware(cache_backend = None, ttl = 60, cache_methods = None, ignore_paths = None, vary_headers = None, cache_control_header = True, compress_responses = False, endpoint_ttls = None, auto_vary = True, invalidation_strategy = None):bases: CacheMiddleware

Advanced middleware for caching API responses.

This middleware extends the basic CacheMiddleware with additional features such as per-endpoint TTL, response compression, and automatic cache key generation.

Initialize the advanced cache middleware.

param cache_backend:

The cache backend to use (defaults to MemoryCacheBackend)

param ttl:

Default time-to-live in seconds

param cache_methods:

Set of HTTP methods to cache (defaults to {“GET”})

param ignore_paths:

Set of paths to exclude from caching

param vary_headers:

Set of headers to include in the cache key

param cache_control_header:

Whether to set Cache-Control headers

param compress_responses:

Whether to compress responses before caching

param endpoint_ttls:

Dictionary mapping endpoints to TTL values

param auto_vary:

Whether to automatically determine vary headers

param invalidation_strategy:

Strategy for cache invalidation

apifrom._generate_cache_key(request)

Generate a cache key for a request.

param request:

The request object

returns:

A cache key string

apifrom._get_endpoint_ttl(path)

Get the TTL for an endpoint.

param path:

The endpoint path

returns:

The TTL in seconds

apifrom.clear()

Clear the cache.

apifrom.get_stats()

Get cache statistics.

returns:

A dictionary of cache statistics

apifrom.invalidate(pattern)

Invalidate cache entries matching a pattern.

param pattern:

The pattern to match

apifrom.process_request(request, call_next)

:async:

Process a request and potentially return a cached response.

param request:

The request object

param call_next:

The next middleware function

returns:

The response object

class apifrom.BatchProcessor(batch_size=100, max_wait_time=0.1, process_func=None, auto_process=True, auto_setup=True)

Parameters:

• batch_size (int)

• max_wait_time (float)

• process_func (Optional[BatchFunction])

• auto_process (bool)

• auto_setup (bool)

Processes batches of similar operations.

This class provides a higher-level interface for batch processing, including batch collection, execution, and statistics.

Initialize a batch processor.

param batch_size:

The maximum number of items in a batch

param max_wait_time:

The maximum time to wait for a batch to fill in seconds

param process_func:

A function to process batches

param auto_process:

Whether to automatically process batches when filled

param auto_setup:

Whether to automatically set up the processor when initialized

apifrom._get_or_create_collector(key)

:async:

Get or create a batch collector.

param key:

The collector key

returns:

A batch collector

apifrom.force_process(collector_key='default')

:async:

Force processing of the current batch.

param collector_key:

The key for the batch collector

apifrom.get_all_stats()

Get batch processing statistics for all collectors.

returns:

A dictionary of statistics by collector key

apifrom.get_stats(collector_key='default')

Get batch processing statistics.

param collector_key:

The key for the batch collector

returns:

A dictionary of statistics

apifrom.map(func, items, batch_size=100, worker_count=5)

:staticmethod:

:async:

Map a function over a list of items with batching and parallelism.

param func:

A function to apply to each item

param items:

The items to process

param batch_size:

The maximum number of items to process in a single batch

param worker_count:

The maximum number of parallel workers

returns:

A list of results

apifrom.process(item, collector_key='default', callback=None)

:async:

Process an item through batch processing.

param item:

The item to process

param collector_key:

The key for the batch collector

param callback:

A callback function to call with the batch results

returns:

The result of processing the item

apifrom.process_batch(items, collector_key='default')

:async:

Process a batch of items.

param items:

The items to process

param collector_key:

The key for the batch collector

returns:

The results of processing the items

apifrom.reset_all_stats()

Reset batch processing statistics for all collectors.

apifrom.reset_stats(collector_key='default')

Reset batch processing statistics.

param collector_key:

The key for the batch collector

apifrom.wait_for_all_empty()

:async:

Wait until all batch collectors are empty.

class apifrom.CacheAnalytics

Collects and analyzes cache performance metrics.

This class collects cache performance metrics such as hit rates, latency, and memory usage, and provides tools for analyzing and visualizing this data.

Initialize cache analytics.

apifrom.get_hot_keys(limit=10)

Get the most frequently accessed cache keys.

param limit:

The maximum number of keys to return

returns:

A list of (key, access_count) tuples

apifrom.get_large_keys(limit=10)

Get the largest cache keys by value size.

param limit:

The maximum number of keys to return

returns:

A list of (key, size) tuples

apifrom.print_summary()

Print a summary of the analytics data to the console.

apifrom.record_delete(key, value_size)

Record a cache delete operation.

param key:

The cache key

param value_size:

The size of the cached value in bytes

apifrom.record_error(key)

Record a cache error.

param key:

The cache key

apifrom.record_hit(key, value_size, latency)

Record a cache hit.

param key:

The cache key

param value_size:

The size of the cached value in bytes

param latency:

The cache lookup latency in seconds

apifrom.record_miss(key, latency)

Record a cache miss.

param key:

The cache key

param latency:

The cache lookup latency in seconds

apifrom.record_request()

Record a cache request.

apifrom.record_set(key, value_size, ttl)

Record a cache set operation.

param key:

The cache key

param value_size:

The size of the cached value in bytes

param ttl:

The TTL in seconds

apifrom.reset()

Reset all analytics data.

apifrom.save(file_path)

Save the analytics data to a file.

param file_path:

The path to save the data to

apifrom.to_dict()

Convert the analytics data to a dictionary.

returns:

A dictionary representation of the analytics data

apifrom.to_json(pretty=True)

Convert the analytics data to a JSON string.

param pretty:

Whether to format the JSON with indentation

returns:

A JSON string representation of the analytics data

class apifrom.CacheBackend

Base class for cache backends.

apifrom.clear()

:abstractmethod:

Clear the cache.

apifrom.delete(key)

:abstractmethod:

Delete a value from the cache.

param key:

The cache key

returns:

True if the key was deleted, False otherwise

apifrom.get(key)

:abstractmethod:

Get a value from the cache.

param key:

The cache key

returns:

The cached value, or None if not found

apifrom.get_stats()

:abstractmethod:

Get cache statistics.

returns:

A dictionary of cache statistics

apifrom.set(key, value, ttl=60)

:abstractmethod:

Set a value in the cache.

param key:

The cache key

param value:

The value to cache

param ttl:

Time-to-live in seconds

class apifrom.CacheOptimizer(cache_backend, strategy=None, analytics=None, auto_tune_interval=None, output_dir=None)

Parameters:

• cache_backend (apifrom.middleware.cache_advanced.CacheBackend)

• strategy (Optional[OptimizedCacheStrategy])

• analytics (Optional[CacheAnalytics])

• auto_tune_interval (Optional[int])

• output_dir (Optional[str])

Optimizes caching for high-load scenarios.

This class provides tools for optimizing cache performance, including strategy optimization, monitoring, and auto-tuning.

Initialize a cache optimizer.

param cache_backend:

The cache backend to optimize

param strategy:

The optimization strategy to use

param analytics:

The cache analytics instance to use

param auto_tune_interval:

The interval in seconds at which to auto-tune

param output_dir:

The directory to save analytics data to

apifrom.auto_tune()

Auto-tune caching parameters based on analytics.

apifrom.delete(key)

:async:

Delete a value from the cache.

param key:

The cache key

returns:

True if the value was deleted, False otherwise

apifrom.get(key)

:async:

Get a value from the cache with optimization.

param key:

The cache key

returns:

The cached value, or None if not found

apifrom.get_analytics()

Get the cache analytics.

returns:

The cache analytics instance

apifrom.reset_analytics()

Reset the cache analytics.

apifrom.set(key, value, ttl=60)

:async:

Set a value in the cache with optimization.

param key:

The cache key

param value:

The value to cache

param ttl:

The TTL in seconds

returns:

True if the value was cached, False otherwise

class apifrom.ConnectionPool(factory, settings=None, validate_func=None, close_func=None, metrics=None)

Parameters:

• factory (Callable[[], Any])

• settings (Optional[ConnectionPoolSettings])

• validate_func (Optional[Callable[[Any], bool]])

• close_func (Optional[Callable[[Any], None]])

• metrics (Optional[ConnectionPoolMetrics])

A generic connection pool for managing connections to external resources.

This class provides a pool of connections to external resources, such as databases, APIs, or services, and manages their lifecycle.

Initialize a connection pool.

param factory:

A factory function that creates new connections

param settings:

The connection pool settings

param validate_func:

A function that validates connections

param close_func:

A function that closes connections

param metrics:

A metrics instance for collecting pool metrics

apifrom._close_connection(connection)

:async:

Close a connection.

param connection:

The connection to close

apifrom._create_connection()

:async:

Create a new connection.

returns:

A new connection

apifrom._validate_connection(connection)

:async:

Validate a connection.

param connection:

The connection to validate

returns:

True if the connection is valid, False otherwise

apifrom.acquire()

:async:

Acquire a connection from the pool.

returns:

A connection from the pool

raises TimeoutError:

If no connection could be acquired within the timeout

raises RuntimeError:

If the pool is closed

apifrom.close()

:async:

Close the connection pool.

This method closes all connections in the pool.

apifrom.connection()

:async:

Get a connection from the pool as a context manager.

This method acquires a connection from the pool, yields it, and ensures it is released back to the pool when done.

Yields:

A connection from the pool

apifrom.get_metrics()

Get the connection pool metrics.

returns:

The connection pool metrics

apifrom.initialize()

:async:

Initialize the connection pool.

This method creates the minimum number of connections.

apifrom.release(connection)

:async:

Release a connection back to the pool.

param connection:

The connection to release

ConnectionPoolMiddleware(database_url = None, redis_url = None, settings = None, max_connections = 20, connection_timeout = 5.0, pool_recycle = 300, pool_pre_ping = True, **kwargs):bases: apifrom.middleware.base.BaseMiddleware

Middleware for providing connection pooling for API requests.

This middleware initializes connection pools for databases and other external services, and provides them to API requests through the request state.

Initialize the connection pool middleware.

param database_url:

The database connection URL

param redis_url:

The Redis connection URL

param settings:

Connection pool settings

param max_connections:

Maximum number of connections in the pool

param connection_timeout:

Connection timeout in seconds

param pool_recycle:

Connection recycle time in seconds

param pool_pre_ping:

Whether to ping connections before using them

param **kwargs:

Additional options

apifrom._initialize_database_pool()

:async:

Initialize the database connection pool.

apifrom._initialize_redis_pool()

:async:

Initialize the Redis connection pool.

apifrom.dispatch(request, call_next)

:async:

Dispatch a request, providing connection pools.

param request:

The request to process

param call_next:

The next middleware or route handler

returns:

The response

apifrom.get_metrics()

Get connection pool metrics.

returns:

A dictionary of metrics

apifrom.initialize()

:async:

Initialize the connection pools.

apifrom.print_metrics()

Print connection pool metrics.

apifrom.process_request(request)

:async:

Process a request (required by BaseMiddleware).

param request:

The request to process

returns:

The processed request

apifrom.process_response(response)

:async:

Process a response (required by BaseMiddleware).

param response:

The response to process

returns:

The processed response

apifrom.shutdown()

:async:

Shutdown all connection pools.

HybridEvictionPolicy:bases: CacheEvictionPolicy

Hybrid eviction policy combining multiple factors.

apifrom.select_items_to_evict(items, target_size)

Select items to evict based on a hybrid policy.

param items:

The list of cache items

param target_size:

The target number of items to keep

returns:

A list of items to evict

JSONExporter(collector = None):bases: MetricsExporter

Exports metrics data in JSON format.

Initialize the exporter with a metrics collector.

apifrom.export(pretty=False)

Export metrics as JSON.

param pretty:

Whether to format the JSON with indentation for readability.

returns:

JSON string representation of the metrics.

apifrom.export_to_file(file_path, pretty=True)

Export metrics to a JSON file.

param file_path:

Path to the output file.

param pretty:

Whether to format the JSON with indentation for readability.

LFUEvictionPolicy:bases: CacheEvictionPolicy

Least Frequently Used (LFU) eviction policy.

apifrom.select_items_to_evict(items, target_size)

Select items to evict based on the LFU policy.

param items:

The list of cache items

param target_size:

The target number of items to keep

returns:

A list of items to evict

LRUEvictionPolicy:bases: CacheEvictionPolicy

Least Recently Used (LRU) eviction policy.

apifrom.select_items_to_evict(items, target_size)

Select items to evict based on the LRU policy.

param items:

The list of cache items

param target_size:

The target number of items to keep

returns:

A list of items to evict

LogExporter(collector = None, logger = None, level = logging.INFO):bases: MetricsExporter

Exports metrics to Python’s logging system.

Initialize the log exporter.

param collector:

Metrics collector to export from.

param logger:

Logger to use for logging metrics.

param level:

Logging level to use.

apifrom.export(include_details=False, **kwargs)

Export metrics to logs.

param include_details:

Whether to include detailed metrics information.

apifrom.export_periodic(interval_seconds=60, include_details=False)

Start a background thread to periodically export metrics to logs.

param interval_seconds:

Interval between exports in seconds.

param include_details:

Whether to include detailed metrics information.

MemoryCacheBackend(max_items = 1000, max_size_bytes = 50 * 1024 * 1024, eviction_policy = None):bases: CacheBackend

In-memory cache backend.

Initialize the memory cache backend.

param max_items:

Maximum number of items to store

param max_size_bytes:

Maximum cache size in bytes

param eviction_policy:

The eviction policy to use

apifrom._cleanup_if_needed(force=False)

Clean up expired items if needed.

param force:

Whether to force cleanup regardless of the time since last cleanup

apifrom._evict_if_needed()

Evict items if the cache exceeds its limits.

apifrom.clear()

Clear the cache.

apifrom.delete(key)

Delete a value from the cache.

param key:

The cache key

returns:

True if the key was deleted, False otherwise

apifrom.get(key)

Get a value from the cache.

param key:

The cache key

returns:

The cached value, or None if not found

apifrom.get_stats()

Get cache statistics.

returns:

A dictionary of cache statistics

apifrom.set(key, value, ttl=60)

Set a value in the cache.

param key:

The cache key

param value:

The value to cache

param ttl:

Time-to-live in seconds

class apifrom.MetricsCollector

Collects and manages metrics for the API.

This class provides a centralized way to create, update, and retrieve metrics for monitoring API performance and usage.

Initialize the metrics collector.

apifrom.create_counter(name, description, labels=None)

Create a new counter metric.

apifrom.create_gauge(name, description, labels=None)

Create a new gauge metric.

apifrom.create_histogram(name, description, buckets=None, labels=None)

Create a new histogram metric.

apifrom.create_metric(metric)

Register a new metric with the collector.

param metric:

The metric to register

returns:

The registered metric

apifrom.create_summary(name, description, quantiles=None, labels=None)

Create a new summary metric.

apifrom.get_all_metrics()

Get all metrics.

apifrom.get_metric(name)

Get a metric by name.

apifrom.increment(name, amount=1.0, labels=None)

Increment a counter metric.

apifrom.observe(name, value, labels=None)

Add an observation to a histogram or summary metric.

apifrom.reset()

Reset all metrics.

apifrom.set(name, value, labels=None)

Set a gauge metric to a specific value.

apifrom.start_timer(name)

Start a timer for measuring durations.

param name:

The name of the metric to record the duration

returns:

A unique timer ID

apifrom.stop_timer(timer_id, labels=None)

Stop a timer and return the duration in milliseconds.

param timer_id:

The timer ID

param labels:

Optional labels to attach to the observation

returns:

The duration in milliseconds

apifrom.track_error(error_type, endpoint=None)

Track an API error.

apifrom.track_request(endpoint)

Track the start of an API request.

param endpoint:

The API endpoint being called

returns:

A timer ID for tracking the request duration

apifrom.track_request_end(timer_id, endpoint, status_code)

Track the end of an API request.

param timer_id:

The timer ID returned by track_request

param endpoint:

The API endpoint that was called

param status_code:

The HTTP status code of the response

returns:

The request duration in milliseconds

class apifrom.MetricsMiddleware(collector=None)

Parameters:

collector (Optional[apifrom.monitoring.metrics.MetricsCollector])

Middleware for collecting metrics during API request processing.

This middleware tracks request counts, durations, and error rates for each endpoint in the API.

Initialize the metrics middleware.

param collector:

Metrics collector to use for tracking metrics.

apifrom.on_error(request, error, endpoint)

:async:

Process an error that occurred during request handling.

param request:

The request object.

param error:

The exception that was raised.

param endpoint:

The endpoint being called.

apifrom.post_request(request, response)

:async:

Process a response after it is generated by the endpoint.

param request:

The request object.

param response:

The response object.

apifrom.pre_request(request, endpoint)

:async:

Process a request before it is handled by the endpoint.

param request:

The request object.

param endpoint:

The endpoint being called.

apifrom.register(app)

:abstractmethod:

Register the middleware with the application.

param app:

The application to register with.

Middleware(app, dispatch = None):bases: starlette.middleware.base.BaseHTTPMiddleware, BaseMiddleware

Middleware class for APIFromAnything.

This class implements the BaseMiddleware interface and extends Starlette’s BaseHTTPMiddleware to provide a middleware component that can be used with Starlette.

Initialize a new BaseMiddleware instance.

param **options:

Options for the middleware.

apifrom.dispatch(request, call_next)

:async:

Dispatch a request.

This method is called by Starlette to process a request.

param request:

The request to process.

param call_next:

A function to call the next middleware or route handler.

returns:

The response.

apifrom.process_request(request)

:abstractmethod:

:async:

Process a request before it is handled by the API.

param request:

The request to process.

returns:

The processed request.

apifrom.process_response(response)

:async:

Process a response after it is handled by the API.

param response:

The response to process.

returns:

The processed response.

class apifrom.OptimizationAnalyzer(config=None, profiler=None, output_dir=None)

Parameters:

• config (Optional[OptimizationConfig])

• profiler (Optional[apifrom.performance.profiler.APIProfiler])

• output_dir (Optional[str])

Analyzes API performance and generates optimization recommendations.

This class provides tools for analyzing API performance, generating optimization recommendations, and applying optimizations.

Initialize an optimization analyzer.

param config:

The optimization configuration

param profiler:

An existing APIProfiler instance

param output_dir:

The directory to save optimization data to

apifrom._analyze_profile_reports(reports)

Analyze profile reports and generate optimization recommendations.

param reports:

The profile reports to analyze

returns:

A list of optimization recommendations

apifrom._analyze_system_stats()

Analyze system statistics and generate optimization recommendations.

returns:

A list of optimization recommendations

apifrom._generate_code_specific_recommendations()

Generate code-specific optimization recommendations.

This method analyzes the codebase to find potential performance issues.

returns:

A list of optimization recommendations

apifrom._generate_general_recommendations()

Generate general optimization recommendations.

returns:

A list of optimization recommendations

apifrom._get_system_stats()

Get system statistics.

returns:

A dictionary of system statistics

apifrom._save_recommendations()

Save optimization recommendations to files.

apifrom.analyze()

Analyze API performance and generate optimization recommendations.

returns:

A list of optimization recommendations

apifrom.get_optimization_config()

Get the optimization configuration.

returns:

The optimization configuration

apifrom.get_recommendations()

Get optimization recommendations.

returns:

A list of optimization recommendations

apifrom.print_summary()

Print a summary of the optimization analysis to the console.

apifrom.set_optimization_config(config)

Set the optimization configuration.

param config:

The optimization configuration

class apifrom.OptimizationConfig(level=OptimizationLevel.BALANCED, enable_caching=True, enable_connection_pooling=True, enable_profiling=True, enable_request_coalescing=False, enable_request_throttling=False, enable_auto_tuning=True, enable_eager_loading=False, enable_circuit_breaker=False, slow_response_threshold_ms=500, high_memory_threshold_mb=500, high_cpu_threshold_percent=80, high_error_rate_threshold=0.05, optimization_interval=3600)

Parameters:

• level (OptimizationLevel)

• enable_caching (bool)

• enable_connection_pooling (bool)

• enable_profiling (bool)

• enable_request_coalescing (bool)

• enable_request_throttling (bool)

• enable_auto_tuning (bool)

• enable_eager_loading (bool)

• enable_circuit_breaker (bool)

• slow_response_threshold_ms (int)

• high_memory_threshold_mb (int)

• high_cpu_threshold_percent (int)

• high_error_rate_threshold (float)

• optimization_interval (int)

Configuration for optimizing APIFromAnything.

This class defines the configuration for optimizing the APIFromAnything application for high-load scenarios.

Initialize optimization configuration.

param level:

The optimization level

param enable_caching:

Whether to enable caching

param enable_connection_pooling:

Whether to enable connection pooling

param enable_profiling:

Whether to enable profiling

param enable_request_coalescing:

Whether to enable request coalescing

param enable_request_throttling:

Whether to enable request throttling

param enable_auto_tuning:

Whether to enable auto-tuning

param enable_eager_loading:

Whether to enable eager loading

param enable_circuit_breaker:

Whether to enable circuit breaker

param slow_response_threshold_ms:

The slow response threshold in milliseconds

param high_memory_threshold_mb:

The high memory threshold in megabytes

param high_cpu_threshold_percent:

The high CPU threshold in percent

param high_error_rate_threshold:

The high error rate threshold (0.0 to 1.0)

param optimization_interval:

The interval between optimizations in seconds

apifrom.create_aggressive()

:classmethod:

Create an aggressive optimization configuration.

These settings are optimized for maximum performance.

returns:

OptimizationConfig instance

apifrom.create_balanced()

:classmethod:

Create a balanced optimization configuration.

These settings provide a balance between performance and overhead.

returns:

OptimizationConfig instance

apifrom.create_minimal()

:classmethod:

Create a minimal optimization configuration.

These settings are optimized for minimal overhead.

returns:

OptimizationConfig instance

apifrom.load(file_path)

:classmethod:

Load configuration from a file.

param file_path:

The path to load the configuration from

returns:

OptimizationConfig instance

apifrom.save(file_path)

Save the configuration to a file.

param file_path:

The path to save the configuration to

apifrom.to_dict()

Convert the configuration to a dictionary.

returns:

A dictionary representation of the configuration

apifrom.to_json(pretty=True)

Convert the configuration to a JSON string.

param pretty:

Whether to format the JSON with indentation

returns:

A JSON string representation of the configuration

OptimizationLevel:bases: enum.Enum

Optimization level for the APIFromAnything application.

This enum defines the level of optimization to apply to the APIFromAnything application.

class apifrom.OptimizationRecommendation(title, description, action, priority=3, affected_endpoints=None, affected_components=None, estimated_impact=None, code_examples=None)

Parameters:

• title (str)

• description (str)

• action (str)

• priority (int)

• affected_endpoints (Optional[List[str]])

• affected_components (Optional[List[str]])

• estimated_impact (Optional[str])

• code_examples (Optional[List[str]])

Represents a performance optimization recommendation.

This class provides methods for generating and applying optimization recommendations based on performance data.

Initialize an optimization recommendation.

param title:

The recommendation title

param description:

The recommendation description

param action:

The recommended action

param priority:

The recommendation priority (1-5, 1 is highest)

param affected_endpoints:

The affected API endpoints

param affected_components:

The affected components

param estimated_impact:

The estimated impact of the recommendation

param code_examples:

Code examples illustrating the recommendation

apifrom.print_summary()

Print a summary of the recommendation to the console.

apifrom.save(file_path)

Save the recommendation to a file.

param file_path:

The path to save the recommendation to

apifrom.to_dict()

Convert the recommendation to a dictionary.

returns:

A dictionary representation of the recommendation

apifrom.to_json(pretty=True)

Convert the recommendation to a JSON string.

param pretty:

Whether to format the JSON with indentation

returns:

A JSON string representation of the recommendation

OptimizedCacheMiddleware(cache_backend=None, ttl = 60, auto_tune = True, auto_tune_interval = 300, output_dir = None, compression = True, **options):bases: apifrom.middleware.base.BaseMiddleware

Middleware for adding optimized caching to an API.

This middleware adds optimized caching to an API, using the CacheOptimizer to optimize caching strategies based on request patterns.

Initialize the optimized cache middleware.

param cache_backend:

The cache backend to use

param ttl:

Default time-to-live in seconds

param auto_tune:

Whether to auto-tune cache parameters

param auto_tune_interval:

The interval to auto-tune cache parameters in seconds

param output_dir:

The directory to save analytics to

param compression:

Whether to compress cached responses

param **options:

Additional options

apifrom._auto_tune()

:async:

Auto-tune cache parameters based on analytics.

apifrom._cache_response(cache_key, response)

:async:

Cache a response.

param cache_key:

The cache key

param response:

The response to cache

apifrom._generate_cache_key(request)

Generate a cache key for a request.

param request:

The request to generate a key for

returns:

The cache key

apifrom._get_from_cache(cache_key)

:async:

Get a response from the cache.

param cache_key:

The cache key

returns:

The cached response, or None if not found

apifrom._is_cacheable(request, response)

Check if a response is cacheable.

param request:

The request

param response:

The response

returns:

True if the response is cacheable, False otherwise

apifrom.clear_cache()

Clear the cache.

apifrom.dispatch(request, call_next)

:async:

Process a request through the middleware.

param request:

The request to process

param call_next:

The next middleware in the chain

returns:

The response from the next middleware

apifrom.get_analytics()

Get cache analytics.

returns:

The cache analytics

apifrom.get_stats()

Get cache statistics.

returns:

A dictionary of cache statistics

apifrom.process_request(request)

:async:

Process a request (required by BaseMiddleware).

param request:

The request to process

returns:

The processed request

apifrom.process_response(response)

:async:

Process a response (required by BaseMiddleware).

param response:

The response to process

returns:

The processed response

apifrom.save_analytics()

Save cache analytics to a file.

returns:

The path to the saved file, or None if saving failed

class apifrom.PoolManager

Manages multiple connection pools.

This class manages multiple connection pools for different resources, such as databases, APIs, or services.

Initialize a pool manager.

apifrom.close_all()

:async:

Close all connection pools.

apifrom.close_pool(name)

:async:

Close a connection pool.

param name:

The name of the pool

apifrom.create_pool(name, factory, settings=None, validate_func=None, close_func=None)

:async:

Create a new connection pool.

param name:

The name of the pool

param factory:

A factory function that creates new connections

param settings:

The connection pool settings

param validate_func:

A function that validates connections

param close_func:

A function that closes connections

returns:

The created connection pool

apifrom.get_metrics()

Get metrics for all connection pools.

returns:

A dictionary mapping pool names to metrics

apifrom.get_pool(name)

:async:

Get a connection pool by name.

param name:

The name of the pool

returns:

The connection pool, or None if not found

apifrom.print_summary()

Print a summary of all connection pools.

ProfileMiddleware(output_dir = None, save_interval = 300, enabled = True, profiler = None):bases: apifrom.middleware.base.BaseMiddleware

Middleware for profiling API requests and responses.

This middleware profiles API requests and responses, collecting performance metrics and generating profile reports. It can be used to identify performance bottlenecks and optimize API performance.

Initialize the profile middleware.

param output_dir:

The directory to save profile reports to

param save_interval:

The interval to save profile reports in seconds

param enabled:

Whether profiling is enabled

param profiler:

The profiler to use (creates a new one if None)

apifrom.clear()

Clear all profile data.

apifrom.disable()

Disable profiling.

apifrom.dispatch(request, call_next)

:async:

Dispatch a request, profiling the execution time.

param request:

The request to process

param call_next:

The next middleware or route handler

returns:

The response

apifrom.enable()

Enable profiling.

apifrom.get_all_reports()

Get all profile reports.

returns:

A list of profile reports

apifrom.get_report(endpoint)

Get a profile report for an endpoint.

param endpoint:

The endpoint to get the report for

returns:

The profile report, or None if not found

apifrom.process_request(request)

:async:

Process a request (required by BaseMiddleware).

param request:

The request to process

returns:

The processed request

apifrom.process_response(response)

:async:

Process a response (required by BaseMiddleware).

param response:

The response to process

returns:

The processed response

class apifrom.ProfileReport(endpoint_name, profile_data)

Parameters:

• endpoint_name (str)

• profile_data (Dict[str, Any])

Represents a performance profile report for an API endpoint.

This class provides methods for analyzing and visualizing profile data, as well as exporting it to various formats.

Initialize a profile report.

param endpoint_name:

The name of the endpoint being profiled

param profile_data:

The raw profile data

apifrom.get_recommendations()

Get performance optimization recommendations based on the profile data.

returns:

A list of recommendations

apifrom.print_summary()

Print a summary of the report to the console.

apifrom.save(file_path)

Save the report to a file.

param file_path:

The path to save the report to

apifrom.to_dict()

Convert the report to a dictionary.

returns:

A dictionary representation of the report

apifrom.to_json(pretty=True)

Convert the report to a JSON string.

param pretty:

Whether to format the JSON with indentation

returns:

A JSON string representation of the report

PrometheusExporter(collector = None):bases: MetricsExporter

Exports metrics in Prometheus text format.

Initialize the exporter with a metrics collector.

apifrom._format_labels(labels)

Format labels for Prometheus exposition format.

apifrom.export(**kwargs)

Export metrics in Prometheus text format.

returns:

String in Prometheus exposition format.

apifrom.export_to_file(file_path)

Export metrics to a file in Prometheus text format.

param file_path:

Path to the output file.

apifrom.serve_metrics(host='localhost', port=8000, endpoint='/metrics')

Start a simple HTTP server to expose metrics for Prometheus scraping.

param host:

Host to bind the server to.

param port:

Port to listen on.

param endpoint:

URL endpoint for metrics.

RedisCacheBackend(redis_url = 'redis://localhost:6379/0', prefix = 'apifrom:', serializer = None, deserializer = None):bases: CacheBackend

Redis cache backend.

Initialize the Redis cache backend.

param redis_url:

The Redis connection URL

param prefix:

The key prefix to use

param serializer:

Function to serialize values to bytes

param deserializer:

Function to deserialize bytes to values

apifrom._make_key(key)

Create a prefixed key.

param key:

The original key

returns:

The prefixed key

apifrom.clear()

Clear the cache.

apifrom.delete(key)

Delete a value from the cache.

param key:

The cache key

returns:

True if the key was deleted, False otherwise

apifrom.get(key)

Get a value from the cache.

param key:

The cache key

returns:

The cached value, or None if not found

apifrom.get_stats()

Get cache statistics.

returns:

A dictionary of cache statistics

apifrom.set(key, value, ttl=60)

Set a value in the cache.

param key:

The cache key

param value:

The value to cache

param ttl:

Time-to-live in seconds

class apifrom.RequestCoalescingMiddleware(window_time=0.1, max_requests=None)

Parameters:

• window_time (float)

• max_requests (Optional[int])

Middleware for coalescing requests.

This middleware can be used to coalesce requests at the middleware level.

Initialize the request coalescing middleware.

param window_time:

Time window in seconds to coalesce requests.

type window_time:

float

param max_requests:

Maximum number of requests to coalesce.

type max_requests:

int, optional

apifrom.get_stats()

Get statistics about the middleware.

returns:

Statistics about the middleware.

rtype:

dict

apifrom.process_request(request)

Process a request, potentially coalescing it with other identical requests.

param request:

Request to process.

type request:

Any

returns:

Processed request.

rtype:

Any

apifrom.process_response(request, response)

Process a response.

param request:

Request that generated the response.

type request:

Any

param response:

Response to process.

type response:

Any

returns:

Processed response.

rtype:

Any

SizeEvictionPolicy:bases: CacheEvictionPolicy

Size-based eviction policy.

apifrom.select_items_to_evict(items, target_size)

Select items to evict based on size.

param items:

The list of cache items

param target_size:

The target number of items to keep

returns:

A list of items to evict

TTLEvictionPolicy:bases: CacheEvictionPolicy

Time-To-Live (TTL) eviction policy.

apifrom.select_items_to_evict(items, target_size)

Select items to evict based on the TTL policy.

param items:

The list of cache items

param target_size:

The target number of items to keep

returns:

A list of items to evict

class apifrom.Web

Decorator for optimizing web endpoints.

This decorator applies various optimizations to web endpoints, such as caching, profiling, and auto-tuning.

apifrom.create_default_web_optimization()

:staticmethod:

Create a default web optimization decorator.

returns:

A default web optimization decorator with balanced settings

apifrom.get_optimization_config(func)

:staticmethod:

Get the optimization configuration for a function.

param func:

The function to get the configuration for

returns:

The optimization configuration dictionary, or an empty dictionary if not optimized

apifrom.is_optimized(func)

:staticmethod:

Check if a function is optimized.

param func:

The function to check

returns:

True if the function is optimized, False otherwise

apifrom.optimize(cache_ttl=None, profile=True, connection_pool=True, auto_tune=True, request_coalescing=False, coalescing_ttl=30, batch_processing=False, batch_size=100, request_throttling=False, eager_loading=False, circuit_breaker=False)

:staticmethod:

Decorator for optimizing web endpoints.

param cache_ttl:

The cache TTL in seconds (None for no caching)

param profile:

Whether to profile the endpoint

param connection_pool:

Whether to use connection pooling

param auto_tune:

Whether to auto-tune the endpoint

param request_coalescing:

Whether to coalesce duplicate requests

param coalescing_ttl:

The time-to-live for coalesced requests in seconds

param batch_processing:

Whether to enable batch processing

param batch_size:

The maximum batch size for batch processing

param request_throttling:

Whether to throttle excessive requests

param eager_loading:

Whether to eagerly load related resources

param circuit_breaker:

Whether to use a circuit breaker

returns:

A decorated function

Functions

apifrom.api(route=None, method='GET', name=None, description=None, tags=None, response_model=None, status_code=200, deprecated=False, include_in_schema=True, **kwargs)

Parameters:

• route (str)

• method (str)

• name (str)

• description (str)

• tags (List[str])

• response_model (Type)

• status_code (int)

• deprecated (bool)

• include_in_schema (bool)

Decorator to expose a Python function as an API endpoint.

param route:

The route for the endpoint. If None, derived from function name.

param method:

The HTTP method for the endpoint.

param name:

The name for the endpoint. If None, derived from function name.

param description:

The description for the endpoint.

param tags:

Tags for the endpoint.

param response_model:

The response model for the endpoint.

param status_code:

The default status code for successful responses.

param deprecated:

Whether the endpoint is deprecated.

param include_in_schema:

Whether to include the endpoint in the API schema.

param **kwargs:

Additional arguments to pass to the router.

returns:

A decorator function.

apifrom.api_key_required(func=None, *, api_keys=None, scopes=None, error_message='Invalid or missing API key')

Parameters:

• api_keys (Optional[Dict[str, Union[str, List[str], Dict[str, Any]]]])

• scopes (Optional[List[str]])

• error_message (str)

Decorator to require a valid API key for an API endpoint.

param func:

The function to decorate.

param api_keys:

A dictionary of API keys and their scopes. If None, uses the API instance’s API keys. The values can be strings, lists of strings, or dictionaries with a ‘scopes’ key.

param scopes:

A list of scopes that the API key must have.

param error_message:

The error message to return if the API key is invalid.

returns:

The decorated function.

apifrom.basic_auth_required(func=None, *, credentials=None, error_message='Invalid or missing credentials')

Parameters:

• credentials (Optional[Dict[str, str]])

• error_message (str)

Decorator to require valid Basic auth credentials for an API endpoint.

param func:

The function to decorate.

param credentials:

A dictionary of username-password pairs. If None, uses the API instance’s Basic auth credentials.

param error_message:

The error message to return if the credentials are invalid.

returns:

The decorated function.

apifrom.batch_process(batch_size=100, max_wait_time=0.1, process_func=None, auto_process=True)

Parameters:

• batch_size (int)

• max_wait_time (float)

• process_func (Optional[BatchFunction])

• auto_process (bool)

Decorator for batch processing.

This decorator groups calls to a function into batches and processes them together.

param batch_size:

The maximum number of items in a batch

param max_wait_time:

The maximum time to wait for a batch to fill in seconds

param process_func:

A function to process batches

param auto_process:

Whether to automatically process batches when filled

returns:

A decorator function

apifrom.coalesce_requests(window_time=0.1, max_requests=None)

Parameters:

• window_time (float)

• max_requests (Optional[int])

Return type:

Callable[[Callable[Ellipsis, R]], Callable[Ellipsis, R]]

Decorator for coalescing multiple identical requests into a single request.

param window_time:

Time window in seconds to coalesce requests.

type window_time:

float

param max_requests:

Maximum number of requests to coalesce.

type max_requests:

int, optional

returns:

Decorated function that coalesces requests.

rtype:

Callable

apifrom.jwt_required(func=None, *, secret=None, algorithm=None, verify_exp=True, verify_aud=False, audience=None, verify_iss=False, issuer=None, verify_sub=False, subject=None, required_claims=None, optional_claims=None, error_message='Invalid or missing JWT token')

Parameters:

• secret (Optional[str])

• algorithm (Optional[str])

• verify_exp (bool)

• verify_aud (bool)

• audience (Optional[str])

• verify_iss (bool)

• issuer (Optional[str])

• verify_sub (bool)

• subject (Optional[str])

• required_claims (Optional[List[str]])

• optional_claims (Optional[List[str]])

• error_message (str)

Decorator that requires a valid JWT token for accessing the endpoint.

param secret:

The secret key used to decode the JWT token

param algorithm:

The algorithm used to decode the JWT token

param verify_exp:

Whether to verify the expiration time

param verify_aud:

Whether to verify the audience

param audience:

The expected audience

param verify_iss:

Whether to verify the issuer

param issuer:

The expected issuer

param verify_sub:

Whether to verify the subject

param subject:

The expected subject

param required_claims:

List of claims that must be present in the token

param optional_claims:

List of claims that may be present in the token

param error_message:

The error message to return if the token is invalid

returns:

The decorated function

apifrom.oauth2_required(func=None, *, scopes=None, token_url=None, error_message='Invalid or missing OAuth2 token')

Parameters:

• scopes (Optional[List[str]])

• token_url (Optional[str])

• error_message (str)

Decorator to require a valid OAuth2 token for an API endpoint.

This is a placeholder implementation. In a real application, you would integrate with an OAuth2 provider like Auth0, Okta, or your own OAuth2 server.

param func:

The function to decorate.

param scopes:

A list of scopes that the token must have.

param token_url:

The URL for obtaining tokens.

param error_message:

The error message to return if the token is invalid.

returns:

The decorated function.