Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .pre-commit-config.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -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 ]
Expand Down
15 changes: 8 additions & 7 deletions examples/scenario_clean_execution.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 ---
Expand All @@ -50,26 +51,26 @@
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()
return infra_bloat


@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)
return data


@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.
Expand All @@ -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)
Expand Down
15 changes: 8 additions & 7 deletions examples/scenario_forgotten_cache.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 ---
Expand All @@ -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
Expand All @@ -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:
Expand All @@ -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
Expand All @@ -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
Expand Down
4 changes: 2 additions & 2 deletions ramflow/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
17 changes: 8 additions & 9 deletions ramflow/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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.
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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)
Expand Down
14 changes: 8 additions & 6 deletions ramflow/reporting.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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,
Expand All @@ -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
Expand Down
Loading