Skip to content

Commit 961dfa7

Browse files
author
matdev83
committed
Manual Merge PR #652 
1 parent 8125449 commit 961dfa7

File tree

7 files changed

+1678
-1565
lines changed

7 files changed

+1678
-1565
lines changed

‎data/test_suite_state.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
11
{
2-
"test_count": 5065,
2+
"test_count": 5068,
33
"last_updated": "1762168167.0802596"
44
}

‎src/constants.py‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1 +1,7 @@
1-
DEFAULT_COMMAND_PREFIX: str = "!/"
1+
"""Project-wide constants used across multiple modules."""
2+
3+
DEFAULT_COMMAND_PREFIX: str = "!/"
4+
"""Default command prefix for interactive commands."""
5+
6+
MAX_RECENT_USAGE_RECORDS: int = 1000
7+
"""Maximum number of recent usage records that can be requested at once."""

‎src/core/app/controllers/usage_controller.py‎

Lines changed: 110 additions & 110 deletions
Original file line numberDiff line numberDiff line change
@@ -1,110 +1,110 @@
1-
"""
2-
Usage controller for exposing usage tracking endpoints.
3-
"""
4-
5-
from __future__ import annotations
6-
7-
import logging
8-
from typing import Any
9-
10-
from fastapi import APIRouter, Depends, Query
11-
12-
from src.core.di.services import get_or_build_service_provider
13-
from src.core.domain.usage_data import UsageData
14-
from src.core.interfaces.usage_tracking_interface import IUsageTrackingService
15-
16-
logger = logging.getLogger(__name__)
17-
18-
router = APIRouter(prefix="/usage", tags=["usage"])
19-
20-
21-
class UsageController:
22-
"""Controller for usage tracking endpoints."""
23-
24-
def __init__(self, usage_service: IUsageTrackingService | None = None) -> None:
25-
"""Initialize the usage controller.
26-
27-
Args:
28-
usage_service: Optional usage tracking service
29-
"""
30-
self.usage_service = usage_service
31-
32-
async def get_usage_stats(
33-
self, project: str | None = None, days: int = 30
34-
) -> dict[str, Any]:
35-
"""Get usage statistics.
36-
37-
Args:
38-
project: Optional project filter
39-
days: Number of days to include in stats
40-
41-
Returns:
42-
Usage statistics dictionary
43-
"""
44-
if not self.usage_service:
45-
return {"error": "Usage tracking service not available"}
46-
47-
result = await self.usage_service.get_usage_stats(project=project, days=days)
48-
return result # type: ignore[no-any-return]
49-
50-
async def get_recent_usage(
51-
self, session_id: str | None = None, limit: int = 100
52-
) -> list[UsageData]:
53-
"""Get recent usage data.
54-
55-
Args:
56-
session_id: Optional session ID filter
57-
limit: Maximum number of records to return
58-
59-
Returns:
60-
List of usage data entities
61-
"""
62-
if not self.usage_service:
63-
return []
64-
65-
result = await self.usage_service.get_recent_usage(
66-
session_id=session_id, limit=limit
67-
)
68-
return result # type: ignore[no-any-return]
69-
70-
71-
@router.get("/stats", response_model=dict[str, Any])
72-
async def get_usage_stats(
73-
project: str | None = Query(None, description="Filter by project name"),
74-
days: int = Query(30, description="Number of days to include in stats"),
75-
service_provider: Any = Depends(get_or_build_service_provider),
76-
) -> dict[str, Any]:
77-
"""Get usage statistics.
78-
79-
Args:
80-
project: Optional project filter
81-
days: Number of days to include in stats
82-
service_provider: Service provider dependency
83-
84-
Returns:
85-
Usage statistics dictionary
86-
"""
87-
usage_service = service_provider.get_required_service(IUsageTrackingService)
88-
result = await usage_service.get_usage_stats(project=project, days=days)
89-
return result # type: ignore[no-any-return]
90-
91-
92-
@router.get("/recent", response_model=list[UsageData])
93-
async def get_recent_usage(
94-
session_id: str | None = Query(None, description="Filter by session ID"),
95-
limit: int = Query(100, description="Maximum number of records to return"),
96-
service_provider: Any = Depends(get_or_build_service_provider),
97-
) -> list[UsageData]:
98-
"""Get recent usage data.
99-
100-
Args:
101-
session_id: Optional session ID filter
102-
limit: Maximum number of records to return
103-
service_provider: Service provider dependency
104-
105-
Returns:
106-
List of usage data entities
107-
"""
108-
usage_service = service_provider.get_required_service(IUsageTrackingService)
109-
result = await usage_service.get_recent_usage(session_id=session_id, limit=limit)
110-
return result # type: ignore[no-any-return]
1+
"""
2+
Usage controller for exposing usage tracking endpoints.
3+
"""
4+
5+
from __future__ import annotations
6+
7+
import logging
8+
from typing import Any
9+
10+
from fastapi import APIRouter, Depends, Query
11+
12+
from src.core.di.services import get_or_build_service_provider
13+
from src.core.domain.usage_data import UsageData
14+
from src.core.interfaces.usage_tracking_interface import IUsageTrackingService
15+
16+
logger = logging.getLogger(__name__)
17+
18+
router = APIRouter(prefix="/usage", tags=["usage"])
19+
20+
21+
class UsageController:
22+
"""Controller for usage tracking endpoints."""
23+
24+
def __init__(self, usage_service: IUsageTrackingService | None = None) -> None:
25+
"""Initialize the usage controller.
26+
27+
Args:
28+
usage_service: Optional usage tracking service
29+
"""
30+
self.usage_service = usage_service
31+
32+
async def get_usage_stats(
33+
self, project: str | None = None, days: int = 30
34+
) -> dict[str, Any]:
35+
"""Get usage statistics.
36+
37+
Args:
38+
project: Optional project filter
39+
days: Number of days to include in stats
40+
41+
Returns:
42+
Usage statistics dictionary
43+
"""
44+
if not self.usage_service:
45+
return {"error": "Usage tracking service not available"}
46+
47+
result = await self.usage_service.get_usage_stats(project=project, days=days)
48+
return result # type: ignore[no-any-return]
49+
50+
async def get_recent_usage(
51+
self, session_id: str | None = None, limit: int = 100
52+
) -> list[UsageData]:
53+
"""Get recent usage data.
54+
55+
Args:
56+
session_id: Optional session ID filter
57+
limit: Maximum number of records to return
58+
59+
Returns:
60+
List of usage data entities
61+
"""
62+
if not self.usage_service:
63+
return []
64+
65+
result = await self.usage_service.get_recent_usage(
66+
session_id=session_id, limit=limit
67+
)
68+
return result # type: ignore[no-any-return]
69+
70+
71+
@router.get("/stats", response_model=dict[str, Any])
72+
async def get_usage_stats(
73+
project: str | None = Query(None, description="Filter by project name"),
74+
days: int = Query(30, description="Number of days to include in stats"),
75+
service_provider: Any = Depends(get_or_build_service_provider),
76+
) -> dict[str, Any]:
77+
"""Get usage statistics.
78+
79+
Args:
80+
project: Optional project filter
81+
days: Number of days to include in stats
82+
service_provider: Service provider dependency
83+
84+
Returns:
85+
Usage statistics dictionary
86+
"""
87+
usage_service = service_provider.get_required_service(IUsageTrackingService)
88+
result = await usage_service.get_usage_stats(project=project, days=days)
89+
return result # type: ignore[no-any-return]
90+
91+
92+
@router.get("/recent", response_model=list[UsageData])
93+
async def get_recent_usage(
94+
session_id: str | None = Query(None, description="Filter by session ID"),
95+
limit: int = Query(100, description="Maximum number of records to return"),
96+
service_provider: Any = Depends(get_or_build_service_provider),
97+
) -> list[UsageData]:
98+
"""Get recent usage data.
99+
100+
Args:
101+
session_id: Optional session ID filter
102+
limit: Maximum number of records to return
103+
service_provider: Service provider dependency
104+
105+
Returns:
106+
List of usage data entities
107+
"""
108+
usage_service = service_provider.get_required_service(IUsageTrackingService)
109+
result = await usage_service.get_recent_usage(session_id=session_id, limit=limit)
110+
return result # type: ignore[no-any-return]

‎src/core/common/usage_limits.py‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
"""Utilities for normalizing usage-related request parameters."""
2+
3+
from __future__ import annotations
4+
5+
from typing import Any
6+
7+
from src.constants import MAX_RECENT_USAGE_RECORDS
8+
9+
10+
def normalize_recent_usage_limit(limit: Any) -> int:
11+
"""Normalize the recent usage limit value to a safe, bounded integer.
12+
13+
Args:
14+
limit: The requested limit value that may come from untrusted sources.
15+
16+
Returns:
17+
A non-negative integer that does not exceed :data:`MAX_RECENT_USAGE_RECORDS`.
18+
Invalid or non-positive values yield ``0`` so callers can short-circuit expensive
19+
repository lookups.
20+
"""
21+
22+
try:
23+
numeric_limit = int(limit)
24+
except (TypeError, ValueError):
25+
return 0
26+
27+
if numeric_limit <= 0:
28+
return 0
29+
30+
return min(numeric_limit, MAX_RECENT_USAGE_RECORDS)

0 commit comments

Comments
 (0)