diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 826875a..a16976d 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -14,7 +14,7 @@ repos: - id: check-added-large-files - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.15.14 + rev: v0.16.0 hooks: - id: ruff-check args: [ --fix, --exit-non-zero-on-fix ] diff --git a/examples/scenario_clean_execution.py b/examples/scenario_clean_execution.py index 5fd1c7e..e44a413 100644 --- a/examples/scenario_clean_execution.py +++ b/examples/scenario_clean_execution.py @@ -28,9 +28,10 @@ to its baseline. The aircraft is empty and ready for the next flight. """ -import time import logging -from typing import List, Dict, Any +import time +from typing import Any + from ramflow import tracker # --- LOGGER CONFIGURATION --- @@ -50,10 +51,10 @@ tracker.threshold = 100 # Simulated cache that we WILL clear this time -SEARCH_CACHE: Dict[str, Any] = {} +SEARCH_CACHE: dict[str, Any] = {} -def simulate_enterprise_infrastructure_load() -> List[bytearray]: +def simulate_enterprise_infrastructure_load() -> list[bytearray]: """Simulates the baseline load of a large enterprise project.""" infra_bloat = [bytearray(1024 * 1024) for _ in range(400)] tracker.log_django_bootstrap() @@ -61,7 +62,7 @@ def simulate_enterprise_infrastructure_load() -> List[bytearray]: @tracker.track("Oracle: Data Extraction") -def oracledb_extraction() -> List[bytearray]: +def oracledb_extraction() -> list[bytearray]: """Extracts data and ensures it stays local to the caller.""" data = [bytearray(1024 * 512) for _ in range(300)] time.sleep(1.0) @@ -69,7 +70,7 @@ def oracledb_extraction() -> List[bytearray]: @tracker.track("Elastic: Optimized Check") -def elastic_integrity_check(erp_data: List[bytearray]) -> List[str]: +def elastic_integrity_check(erp_data: list[bytearray]) -> list[str]: """Queries ES with proper filtering and temporary caching. The results are stored globally but will be cleared by the orchestrator. @@ -85,7 +86,7 @@ def elastic_integrity_check(erp_data: List[bytearray]) -> List[str]: @tracker.track("Faktory: Efficient Dispatch") -def slow_dispatch(skus: List[str]) -> None: +def slow_dispatch(skus: list[str]) -> None: """Enqueues tasks while other resources are being freed.""" for i in range(min(len(skus), 10)): time.sleep(0.1) diff --git a/examples/scenario_forgotten_cache.py b/examples/scenario_forgotten_cache.py index bcf2dc9..e3e8498 100644 --- a/examples/scenario_forgotten_cache.py +++ b/examples/scenario_forgotten_cache.py @@ -27,9 +27,10 @@ a massive footprint. On a server running this repeatedly, a crash is inevitable. """ -import time import logging -from typing import List, Dict, Any +import time +from typing import Any + from ramflow import tracker # --- LOGGER CONFIGURATION --- @@ -49,10 +50,10 @@ tracker.threshold = 100 # Threshold in MB for task highlighting # Persistent storage simulation (The "Hidden Culprit") -SEARCH_CACHE: Dict[str, Any] = {} +SEARCH_CACHE: dict[str, Any] = {} -def simulate_enterprise_infrastructure_load() -> List[bytearray]: +def simulate_enterprise_infrastructure_load() -> list[bytearray]: """Simulates a heavy infrastructure load (spaCy, Pandas, Models). This helper function forces a memory allocation to simulate the resident @@ -68,7 +69,7 @@ def simulate_enterprise_infrastructure_load() -> List[bytearray]: @tracker.track("Oracle: Data Extraction ERP") -def oracledb_extraction() -> List[bytearray]: +def oracledb_extraction() -> list[bytearray]: """Simulates pulling heavy product data from a legacy ERP database. Returns: @@ -81,7 +82,7 @@ def oracledb_extraction() -> List[bytearray]: @tracker.track("Elastic: Integrity Check") -def elastic_integrity_check(erp_data: List[bytearray]) -> List[str]: +def elastic_integrity_check(erp_data: list[bytearray]) -> list[str]: """Queries ElasticSearch to validate documents before dispatching. This function intentionally omits '_source' filtering to simulate @@ -107,7 +108,7 @@ def elastic_integrity_check(erp_data: List[bytearray]) -> List[str]: @tracker.track("Faktory: Slow Queue Dispatch") -def slow_dispatch(skus: List[str]) -> None: +def slow_dispatch(skus: list[str]) -> None: """Enqueues tasks one by one to the background worker server. This slow loop mimics network latency, keeping all previously allocated diff --git a/ramflow/__init__.py b/ramflow/__init__.py index 832b167..8a338fc 100644 --- a/ramflow/__init__.py +++ b/ramflow/__init__.py @@ -4,10 +4,10 @@ generate visual execution trees, and audit memory release integrity. """ -from .core import tracker, RamFlow +from .core import RamFlow, tracker __version__: str = "1.0.0" -__all__: list[str] = ["tracker", "RamFlow"] +__all__: list[str] = ["RamFlow", "tracker"] def get_tracker() -> RamFlow: diff --git a/ramflow/core.py b/ramflow/core.py index e253d02..b7f7e10 100644 --- a/ramflow/core.py +++ b/ramflow/core.py @@ -6,11 +6,13 @@ import os import time -import psutil +from collections.abc import Callable from datetime import datetime from functools import wraps from pathlib import Path -from typing import Any, Callable, Dict, List, Optional, Union +from typing import Any + +import psutil from .config import HARD_LIMIT, THRESHOLD from .utils import force_release, kill_process @@ -48,7 +50,7 @@ def __init__( self.django_overhead: float = 0.0 self.threshold: int = threshold self.hard_limit: int = hard_limit_mb - self.history: List[Dict[str, Any]] = [] + self.history: list[dict[str, Any]] = [] @property def env(self) -> str: @@ -76,7 +78,6 @@ def log_bootstrap(self) -> None: This method is kept for API consistency. The actual baseline is captured during class initialization. """ - pass def log_django_bootstrap(self) -> None: """Records the memory overhead of the Django framework initialization. @@ -98,7 +99,7 @@ def log_django_bootstrap(self) -> None: } ) - def track(self, label: Optional[Union[str, Callable]] = None) -> Callable: + def track(self, label: str | Callable | None = None) -> Callable: """Decorator to monitor the net memory consumption of a function. Calculates 'Net Self' memory by subtracting the consumption of any @@ -192,9 +193,7 @@ def check_final_release(self, kill_on_leak: bool = False) -> float: return leak - def generate_report( - self, folder: Optional[str] = None, suffix: str = "audit" - ) -> str: + def generate_report(self, folder: str | None = None, suffix: str = "audit") -> str: """Generates the Premium HTML report with automated smart naming. Triggers a final release audit, instantiates the reporter, and saves @@ -207,8 +206,8 @@ def generate_report( Returns: str: The absolute path to the generated HTML report. """ - from .reporting import Reporter from .config import REPORTS_DIR + from .reporting import Reporter target_dir = Path.cwd() / (folder or REPORTS_DIR) target_dir.mkdir(parents=True, exist_ok=True) diff --git a/ramflow/reporting.py b/ramflow/reporting.py index 4487e70..29b90bc 100644 --- a/ramflow/reporting.py +++ b/ramflow/reporting.py @@ -4,14 +4,16 @@ visual execution timelines, and detailed system metadata. """ -import os +import getpass import json -import psutil +import os import platform import socket -import getpass -from typing import List, Dict, Any from datetime import datetime +from typing import Any + +import psutil + from . import config from .templates import components @@ -29,7 +31,7 @@ class Reporter: def __init__( self, - history: List[Dict[str, Any]], + history: list[dict[str, Any]], start_mem: float, env: str, threshold: int, @@ -50,7 +52,7 @@ def __init__( self.threshold = threshold self.django_overhead = django_overhead - def _analyze_memory_trend(self, residual_leak: float) -> Dict[str, str]: + def _analyze_memory_trend(self, residual_leak: float) -> dict[str, str]: """Analyzes cumulative memory pressure to detect persistent bloating. This algorithm evaluates the 'Memory Plateau' effect. If the process