apifrom.performance

Performance optimization modules for APIFromAnything.

This module provides tools for optimizing APIFromAnything applications for high-load scenarios, including performance monitoring, profiling, and optimization.

Overview

Classes

• APIProfiler

• BatchCollector

• BatchExecutor

• BatchProcessor

• CacheAnalytics

• CacheOptimizer

• CoalescedRequest

• ConnectionPool

• ConnectionPoolMetrics

• ConnectionPoolMiddleware

• ConnectionPoolSettings

• OptimizationAnalyzer

• OptimizationConfig

• OptimizationLevel

• OptimizationRecommendation

• OptimizedCacheMiddleware

• OptimizedCacheStrategy

• PoolManager

• ProfileMiddleware

• ProfileReport

• RequestCoalescingManager

• RequestCoalescingMiddleware

• Web

Functions

• batch_process()

• coalesce_requests()

Classes

class apifrom.performance.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.performance.clear()

Clear all profile data.

apifrom.performance.disable()

Disable profiling.

apifrom.performance.enable()

Enable profiling.

apifrom.performance.get_all_reports()

Get profile reports for all endpoints.

returns:

A list of ProfileReport instances

apifrom.performance.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.performance.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.performance.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

BatchCollector(max_batch_size = 100, max_wait_time = 0.1, process_func = None, auto_process = True):bases: Generic[T]

Collects items for batch processing.

This class collects items until a batch size or timeout is reached, then processes them in a batch.

Initialize a batch collector.

param max_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

apifrom.performance._process_batch()

:async:

Process the current batch.

apifrom.performance._start_timer()

Start a timer to process the batch after max_wait_time seconds.

apifrom.performance.add_item(item, callback=None)

:async:

Add an item to the batch and return a future that will resolve when the item is processed.

param item:

The item to add to the batch

param callback:

A callback function to call with the batch results

returns:

A future that will resolve when the item is processed

apifrom.performance.get_stats()

Get batch processing statistics.

returns:

A dictionary of statistics

apifrom.performance.process_current_batch()

:async:

Process the current batch immediately.

apifrom.performance.reset_stats()

Reset batch processing statistics.

apifrom.performance.wait_for_empty()

:async:

Wait until the batch collector is empty.

class apifrom.performance.BatchExecutor

Executes batched operations.

This class provides methods for executing batched operations with optimal batch sizes and parallelism.

apifrom.performance.execute_batch(batch_func, items, batch_size=100, parallel=False, max_workers=5)

:staticmethod:

:async:

Execute a batch function on a list of items.

param batch_func:

A function that processes a batch of items

param items:

The items to process

param batch_size:

The maximum number of items to process in a single batch

param parallel:

Whether to process batches in parallel

param max_workers:

The maximum number of parallel workers

returns:

A list of results

apifrom.performance.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.performance.reduce(func, items, initial, batch_size=100)

:staticmethod:

:async:

Reduce a list of items with a function.

param func:

A reduction function

param items:

The items to reduce

param initial:

The initial value

param batch_size:

The maximum number of items to process in a single batch

returns:

The reduced result

class apifrom.performance.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.performance._get_or_create_collector(key)

:async:

Get or create a batch collector.

param key:

The collector key

returns:

A batch collector

apifrom.performance.force_process(collector_key='default')

:async:

Force processing of the current batch.

param collector_key:

The key for the batch collector

apifrom.performance.get_all_stats()

Get batch processing statistics for all collectors.

returns:

A dictionary of statistics by collector key

apifrom.performance.get_stats(collector_key='default')

Get batch processing statistics.

param collector_key:

The key for the batch collector

returns:

A dictionary of statistics

apifrom.performance.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.performance.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.performance.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.performance.reset_all_stats()

Reset batch processing statistics for all collectors.

apifrom.performance.reset_stats(collector_key='default')

Reset batch processing statistics.

param collector_key:

The key for the batch collector

apifrom.performance.wait_for_all_empty()

:async:

Wait until all batch collectors are empty.

class apifrom.performance.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.performance.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.performance.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.performance.print_summary()

Print a summary of the analytics data to the console.

apifrom.performance.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.performance.record_error(key)

Record a cache error.

param key:

The cache key

apifrom.performance.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.performance.record_miss(key, latency)

Record a cache miss.

param key:

The cache key

param latency:

The cache lookup latency in seconds

apifrom.performance.record_request()

Record a cache request.

apifrom.performance.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.performance.reset()

Reset all analytics data.

apifrom.performance.save(file_path)

Save the analytics data to a file.

param file_path:

The path to save the data to

apifrom.performance.to_dict()

Convert the analytics data to a dictionary.

returns:

A dictionary representation of the analytics data

apifrom.performance.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.performance.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.performance.auto_tune()

Auto-tune caching parameters based on analytics.

apifrom.performance.delete(key)

:async:

Delete a value from the cache.

param key:

The cache key

returns:

True if the value was deleted, False otherwise

apifrom.performance.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.performance.get_analytics()

Get the cache analytics.

returns:

The cache analytics instance

apifrom.performance.reset_analytics()

Reset the cache analytics.

apifrom.performance.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.performance.CoalescedRequest(key, func, args, kwargs)

Parameters:

• key (str)

• func (Callable)

• args (tuple)

• kwargs (dict)

Represents a coalesced request.

This class is used to track multiple identical requests that have been coalesced into a single request.

Initialize a coalesced request.

param key:

Cache key for the request.

type key:

str

param func:

Function to execute.

type func:

Callable

param args:

Positional arguments for the function.

type args:

tuple

param kwargs:

Keyword arguments for the function.

type kwargs:

dict

apifrom.performance.execute()

Execute the request.

returns:

Result of executing the function.

rtype:

Any

class apifrom.performance.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.performance._close_connection(connection)

:async:

Close a connection.

param connection:

The connection to close

apifrom.performance._create_connection()

:async:

Create a new connection.

returns:

A new connection

apifrom.performance._validate_connection(connection)

:async:

Validate a connection.

param connection:

The connection to validate

returns:

True if the connection is valid, False otherwise

apifrom.performance.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.performance.close()

:async:

Close the connection pool.

This method closes all connections in the pool.

apifrom.performance.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.performance.get_metrics()

Get the connection pool metrics.

returns:

The connection pool metrics

apifrom.performance.initialize()

:async:

Initialize the connection pool.

This method creates the minimum number of connections.

apifrom.performance.release(connection)

:async:

Release a connection back to the pool.

param connection:

The connection to release

class apifrom.performance.ConnectionPoolMetrics

Collects and analyzes connection pool metrics.

This class collects connection pool metrics such as pool size, connection acquisition times, and utilization, and provides tools for analyzing and visualizing this data.

Initialize connection pool metrics.

apifrom.performance.print_summary()

Print a summary of the metrics data to the console.

apifrom.performance.record_connection_acquire(connection_id, acquisition_time)

Record a connection acquisition.

param connection_id:

The connection ID

param acquisition_time:

The time it took to acquire the connection in seconds

apifrom.performance.record_connection_error()

Record a connection error.

apifrom.performance.record_connection_release(connection_id, release_time)

Record a connection release.

param connection_id:

The connection ID

param release_time:

The time it took to release the connection in seconds

apifrom.performance.record_connection_request()

Record a connection request.

apifrom.performance.record_connection_timeout()

Record a connection timeout.

apifrom.performance.reset()

Reset all metrics data.

apifrom.performance.save(file_path)

Save the metrics data to a file.

param file_path:

The path to save the data to

apifrom.performance.to_dict()

Convert the metrics data to a dictionary.

returns:

A dictionary representation of the metrics data

apifrom.performance.to_json(pretty=True)

Convert the metrics data to a JSON string.

param pretty:

Whether to format the JSON with indentation

returns:

A JSON string representation of the metrics data

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.performance._initialize_database_pool()

:async:

Initialize the database connection pool.

apifrom.performance._initialize_redis_pool()

:async:

Initialize the Redis connection pool.

apifrom.performance.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.performance.get_metrics()

Get connection pool metrics.

returns:

A dictionary of metrics

apifrom.performance.initialize()

:async:

Initialize the connection pools.

apifrom.performance.print_metrics()

Print connection pool metrics.

apifrom.performance.process_request(request)

:async:

Process a request (required by BaseMiddleware).

param request:

The request to process

returns:

The processed request

apifrom.performance.process_response(response)

:async:

Process a response (required by BaseMiddleware).

param response:

The response to process

returns:

The processed response

apifrom.performance.shutdown()

:async:

Shutdown all connection pools.

class apifrom.performance.ConnectionPoolSettings(min_size=1, max_size=10, max_idle=5, max_lifetime=3600, acquire_timeout=10.0, idle_timeout=300.0, retry_limit=3, retry_delay=1.0, validate_on_acquire=True)

Parameters:

• min_size (int)

• max_size (int)

• max_idle (int)

• max_lifetime (int)

• acquire_timeout (float)

• idle_timeout (float)

• retry_limit (int)

• retry_delay (float)

• validate_on_acquire (bool)

Settings for a connection pool.

This class defines the settings for a connection pool, such as pool size, timeout, and retry parameters.

Initialize connection pool settings.

param min_size:

The minimum pool size

param max_size:

The maximum pool size

param max_idle:

The maximum number of idle connections

param max_lifetime:

The maximum connection lifetime in seconds

param acquire_timeout:

The connection acquisition timeout in seconds

param idle_timeout:

The idle connection timeout in seconds

param retry_limit:

The maximum number of connection retries

param retry_delay:

The delay between connection retries in seconds

param validate_on_acquire:

Whether to validate connections on acquisition

apifrom.performance.create_aggressive()

:classmethod:

Create aggressive connection pool settings.

These settings are optimized for high-throughput, low-latency scenarios.

returns:

ConnectionPoolSettings instance

apifrom.performance.create_balanced()

:classmethod:

Create balanced connection pool settings.

These settings provide a balance between performance and resource usage.

returns:

ConnectionPoolSettings instance

apifrom.performance.create_conservative()

:classmethod:

Create conservative connection pool settings.

These settings are optimized for reliability and resource efficiency.

returns:

ConnectionPoolSettings instance

class apifrom.performance.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.performance._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.performance._analyze_system_stats()

Analyze system statistics and generate optimization recommendations.

returns:

A list of optimization recommendations

apifrom.performance._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.performance._generate_general_recommendations()

Generate general optimization recommendations.

returns:

A list of optimization recommendations

apifrom.performance._get_system_stats()

Get system statistics.

returns:

A dictionary of system statistics

apifrom.performance._save_recommendations()

Save optimization recommendations to files.

apifrom.performance.analyze()

Analyze API performance and generate optimization recommendations.

returns:

A list of optimization recommendations

apifrom.performance.get_optimization_config()

Get the optimization configuration.

returns:

The optimization configuration

apifrom.performance.get_recommendations()

Get optimization recommendations.

returns:

A list of optimization recommendations

apifrom.performance.print_summary()

Print a summary of the optimization analysis to the console.

apifrom.performance.set_optimization_config(config)

Set the optimization configuration.

param config:

The optimization configuration

class apifrom.performance.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.performance.create_aggressive()

:classmethod:

Create an aggressive optimization configuration.

These settings are optimized for maximum performance.

returns:

OptimizationConfig instance

apifrom.performance.create_balanced()

:classmethod:

Create a balanced optimization configuration.

These settings provide a balance between performance and overhead.

returns:

OptimizationConfig instance

apifrom.performance.create_minimal()

:classmethod:

Create a minimal optimization configuration.

These settings are optimized for minimal overhead.

returns:

OptimizationConfig instance

apifrom.performance.load(file_path)

:classmethod:

Load configuration from a file.

param file_path:

The path to load the configuration from

returns:

OptimizationConfig instance

apifrom.performance.save(file_path)

Save the configuration to a file.

param file_path:

The path to save the configuration to

apifrom.performance.to_dict()

Convert the configuration to a dictionary.

returns:

A dictionary representation of the configuration

apifrom.performance.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.performance.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.performance.print_summary()

Print a summary of the recommendation to the console.

apifrom.performance.save(file_path)

Save the recommendation to a file.

param file_path:

The path to save the recommendation to

apifrom.performance.to_dict()

Convert the recommendation to a dictionary.

returns:

A dictionary representation of the recommendation

apifrom.performance.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.performance._auto_tune()

:async:

Auto-tune cache parameters based on analytics.

apifrom.performance._cache_response(cache_key, response)

:async:

Cache a response.

param cache_key:

The cache key

param response:

The response to cache

apifrom.performance._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.performance._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.performance._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.performance.clear_cache()

Clear the cache.

apifrom.performance.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.performance.get_analytics()

Get cache analytics.

returns:

The cache analytics

apifrom.performance.get_stats()

Get cache statistics.

returns:

A dictionary of cache statistics

apifrom.performance.process_request(request)

:async:

Process a request (required by BaseMiddleware).

param request:

The request to process

returns:

The processed request

apifrom.performance.process_response(response)

:async:

Process a response (required by BaseMiddleware).

param response:

The response to process

returns:

The processed response

apifrom.performance.save_analytics()

Save cache analytics to a file.

returns:

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

class apifrom.performance.OptimizedCacheStrategy(min_ttl=60, max_ttl=86400, hit_rate_threshold=0.8, compress_values=True, compress_keys=False, prefetch_keys=False, auto_tune=True)

Parameters:

• min_ttl (int)

• max_ttl (int)

• hit_rate_threshold (float)

• compress_values (bool)

• compress_keys (bool)

• prefetch_keys (bool)

• auto_tune (bool)

A cache strategy that optimizes caching based on runtime analytics.

This class provides cache optimization strategies such as TTL adjustment, key compression, and value compression, based on runtime analytics.

Initialize an optimized cache strategy.

param min_ttl:

The minimum TTL in seconds

param max_ttl:

The maximum TTL in seconds

param hit_rate_threshold:

The hit rate threshold for TTL adjustment

param compress_values:

Whether to compress cached values

param compress_keys:

Whether to compress cache keys

param prefetch_keys:

Whether to prefetch related keys

param auto_tune:

Whether to auto-tune caching parameters

apifrom.performance.get_optimized_ttl(key, default_ttl)

Get an optimized TTL for a key based on access patterns.

param key:

The cache key

param default_ttl:

The default TTL in seconds

returns:

The optimized TTL in seconds

apifrom.performance.optimize_key(key)

Optimize a cache key.

param key:

The original cache key

returns:

The optimized cache key

apifrom.performance.optimize_value(value)

Optimize a cached value.

param value:

The original value

returns:

The optimized value

apifrom.performance.record_hit(key)

Record a cache hit for a key.

param key:

The cache key

apifrom.performance.record_miss(key)

Record a cache miss for a key.

param key:

The cache key

apifrom.performance.record_set(key, ttl)

Record a cache set operation.

param key:

The cache key

param ttl:

The TTL in seconds

class apifrom.performance.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.performance.close_all()

:async:

Close all connection pools.

apifrom.performance.close_pool(name)

:async:

Close a connection pool.

param name:

The name of the pool

apifrom.performance.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.performance.get_metrics()

Get metrics for all connection pools.

returns:

A dictionary mapping pool names to metrics

apifrom.performance.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.performance.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.performance.clear()

Clear all profile data.

apifrom.performance.disable()

Disable profiling.

apifrom.performance.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.performance.enable()

Enable profiling.

apifrom.performance.get_all_reports()

Get all profile reports.

returns:

A list of profile reports

apifrom.performance.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.performance.process_request(request)

:async:

Process a request (required by BaseMiddleware).

param request:

The request to process

returns:

The processed request

apifrom.performance.process_response(response)

:async:

Process a response (required by BaseMiddleware).

param response:

The response to process

returns:

The processed response

class apifrom.performance.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.performance.get_recommendations()

Get performance optimization recommendations based on the profile data.

returns:

A list of recommendations

apifrom.performance.print_summary()

Print a summary of the report to the console.

apifrom.performance.save(file_path)

Save the report to a file.

param file_path:

The path to save the report to

apifrom.performance.to_dict()

Convert the report to a dictionary.

returns:

A dictionary representation of the report

apifrom.performance.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

class apifrom.performance.RequestCoalescingManager

Manager for coalescing requests across the application.

This class provides a centralized way to manage request coalescing across the application.

Initialize the request coalescing manager.

apifrom.performance.execute(func, *args, **kwargs)

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

param func:

Function to execute.

type func:

Callable

param *args:

Positional arguments for the function.

param **kwargs:

Keyword arguments for the function.

returns:

Result of executing the function.

rtype:

Any

apifrom.performance.get_coalescer(func, window_time=0.1, max_requests=None)

Get or create a coalescer for a function.

param func:

Function to coalesce.

type func:

Callable

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:

Coalescer for the function.

rtype:

RequestCoalescer

apifrom.performance.get_stats()

Get statistics about the manager.

returns:

Statistics about the manager.

rtype:

dict

class apifrom.performance.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.performance.get_stats()

Get statistics about the middleware.

returns:

Statistics about the middleware.

rtype:

dict

apifrom.performance.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.performance.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

class apifrom.performance.Web

Decorator for optimizing web endpoints.

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

apifrom.performance.create_default_web_optimization()

:staticmethod:

Create a default web optimization decorator.

returns:

A default web optimization decorator with balanced settings

apifrom.performance.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.performance.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.performance.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.performance.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.performance.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