Refactor imports and standardize file naming: update module imports for consistency, align filenames with snake_case convention, and extract DynamicResponse to a dedicated module.

This commit is contained in:
Yûki VACHOT 2025-11-29 02:27:58 +01:00
parent 126d04ab0f
commit bf5d1b890a
22 changed files with 160 additions and 145 deletions

View file

@ -1,7 +1,8 @@
"""API client module for GeoGuessr communication."""
from .client import DynamicResponse, GeoGuessrClient
from .dynamic_response import DynamicResponse
from .endpoints import EndpointBuilder, EndpointInfo, Endpoints
from .geoguessr_client import GeoGuessrClient
__all__ = [
"GeoGuessrClient",

View file

@ -0,0 +1,121 @@
"""
A wrapper for handling API responses with dynamic schema integration.
This module provides a class, `DynamicResponse`, designed to work with API
responses by incorporating schema information from a dynamic schema registry.
It allows for easier interpretation and manipulation of the API response
data and its metadata.
"""
import logging
from typing import Any
from ..monitoring.schema.schema_registry import schema_registry
logger = logging.getLogger(__name__)
class DynamicResponse:
"""
Wrapper for API responses with dynamic schema information.
This class provides methods to access response data with awareness
of the current schema, making it easier for the LLM to understand
and process the data.
"""
def __init__(
self,
data: Any,
endpoint: str,
status_code: int,
response_time_ms: float,
):
self.data = data
self.endpoint = endpoint
self.status_code = status_code
self.response_time_ms = response_time_ms
self._schema = schema_registry.get_schema(endpoint)
@property
def is_success(self) -> bool:
"""Check if the request was successful."""
return 200 <= self.status_code < 300
@property
def schema_description(self) -> str:
"""Get a human-readable description of the response schema."""
return schema_registry.generate_dynamic_description(self.endpoint)
@property
def available_fields(self) -> list[str]:
"""Get list of available fields in this response."""
if self._schema:
return list(self._schema.fields.keys())
if isinstance(self.data, dict):
return list(self.data.keys())
return []
def get_field(self, field_name: str, default: Any = None) -> Any:
"""
Safely get a field from the response data.
Supports nested field access using dot notation (e.g., "user.profile.name")
"""
if not isinstance(self.data, dict):
return default
parts = field_name.split(".")
current = self.data
for part in parts:
if isinstance(current, dict) and part in current:
current = current[part]
else:
return default
return current
def to_dict(self) -> dict:
"""Convert response to a dictionary with metadata."""
return {
"success": self.is_success,
"status_code": self.status_code,
"endpoint": self.endpoint,
"response_time_ms": round(self.response_time_ms, 2),
"data": self.data,
"available_fields": self.available_fields,
}
def summarize(self, max_depth: int = 2) -> dict:
"""
Create a summarized view of the response for LLM context.
This reduces token usage while providing essential information.
"""
def summarize_value(value: Any, depth: int) -> Any:
if depth <= 0:
if isinstance(value, (dict, list)):
return f"<{type(value).__name__} with {len(value)} items>"
return value
if isinstance(value, dict):
return {k: summarize_value(v, depth - 1) for k, v in list(value.items())[:10]}
if isinstance(value, list):
if len(value) == 0:
return []
return [
summarize_value(value[0], depth - 1),
f"... and {len(value) - 1} more items" if len(value) > 1 else None,
]
if isinstance(value, str) and len(value) > 100:
return value[:100] + "..."
return value
return {
"endpoint": self.endpoint,
"status": "success" if self.is_success else "error",
"field_count": len(self.available_fields),
"data_summary": summarize_value(self.data, max_depth),
}

View file

@ -1,129 +1,28 @@
"""
Dynamic HTTP client for GeoGuessr API communication.
Module for GeoGuessr dynamic HTTP client.
This client automatically handles authentication, endpoint routing,
and integrates with the schema registry for dynamic response handling.
The module encapsulates the HTTP client for interacting with the
GeoGuessr API, including features such as authentication handling,
response schema tracking, retry logic, and integrated monitoring.
Classes:
- GeoGuessrClient: The main HTTP client for communicating with the GeoGuessr API.
"""
import logging
from typing import Any, Optional
from typing import Optional
import httpx
from .dynamic_response import DynamicResponse
from .endpoints import EndpointInfo
from ..auth.session import SessionManager
from ..config import settings
from ..monitoring.schema.SchemaRegistry import schema_registry
from ..monitoring.schema.schema_registry import schema_registry
logger = logging.getLogger(__name__)
class DynamicResponse:
"""
Wrapper for API responses with dynamic schema information.
This class provides methods to access response data with awareness
of the current schema, making it easier for the LLM to understand
and process the data.
"""
def __init__(
self,
data: Any,
endpoint: str,
status_code: int,
response_time_ms: float,
):
self.data = data
self.endpoint = endpoint
self.status_code = status_code
self.response_time_ms = response_time_ms
self._schema = schema_registry.get_schema(endpoint)
@property
def is_success(self) -> bool:
"""Check if the request was successful."""
return 200 <= self.status_code < 300
@property
def schema_description(self) -> str:
"""Get a human-readable description of the response schema."""
return schema_registry.generate_dynamic_description(self.endpoint)
@property
def available_fields(self) -> list[str]:
"""Get list of available fields in this response."""
if self._schema:
return list(self._schema.fields.keys())
if isinstance(self.data, dict):
return list(self.data.keys())
return []
def get_field(self, field_name: str, default: Any = None) -> Any:
"""
Safely get a field from the response data.
Supports nested field access using dot notation (e.g., "user.profile.name")
"""
if not isinstance(self.data, dict):
return default
parts = field_name.split(".")
current = self.data
for part in parts:
if isinstance(current, dict) and part in current:
current = current[part]
else:
return default
return current
def to_dict(self) -> dict:
"""Convert response to a dictionary with metadata."""
return {
"success": self.is_success,
"status_code": self.status_code,
"endpoint": self.endpoint,
"response_time_ms": round(self.response_time_ms, 2),
"data": self.data,
"available_fields": self.available_fields,
}
def summarize(self, max_depth: int = 2) -> dict:
"""
Create a summarized view of the response for LLM context.
This reduces token usage while providing essential information.
"""
def summarize_value(value: Any, depth: int) -> Any:
if depth <= 0:
if isinstance(value, (dict, list)):
return f"<{type(value).__name__} with {len(value)} items>"
return value
if isinstance(value, dict):
return {k: summarize_value(v, depth - 1) for k, v in list(value.items())[:10]}
if isinstance(value, list):
if len(value) == 0:
return []
return [
summarize_value(value[0], depth - 1),
f"... and {len(value) - 1} more items" if len(value) > 1 else None,
]
if isinstance(value, str) and len(value) > 100:
return value[:100] + "..."
return value
return {
"endpoint": self.endpoint,
"status": "success" if self.is_success else "error",
"field_count": len(self.available_fields),
"data_summary": summarize_value(self.data, max_depth),
}
class GeoGuessrClient:
"""
Dynamic HTTP client for GeoGuessr API.

View file

@ -1,12 +1,12 @@
"""Data models for GeoGuessr."""
from .Achievement import Achievement
from .DailyChallenge import DailyChallenge
from .Game import Game
from .RoundGuess import RoundGuess
from .SeasonStats import SeasonStats
from .UserProfile import UserProfile
from .UserStats import UserStats
from .achievement import Achievement
from .daily_challenge import DailyChallenge
from .game import Game
from .round_guess import RoundGuess
from .season_stats import SeasonStats
from .user_profile import UserProfile
from .user_stats import UserStats
__all__ = [
"UserProfile",

View file

@ -3,7 +3,7 @@
from dataclasses import dataclass, field
from typing import Optional
from .RoundGuess import RoundGuess
from .round_guess import RoundGuess
@dataclass

View file

@ -1,10 +1,10 @@
"""Monitoring module for API endpoint tracking and schema detection."""
from .endpoint.EndpointMonitor import (MONITORED_ENDPOINTS, EndpointMonitor,
from .endpoint.endpoint_monitor import (MONITORED_ENDPOINTS, EndpointMonitor,
endpoint_monitor)
from .schema.EndpointSchema import EndpointSchema
from .schema.SchemaDetector import SchemaDetector, SchemaField
from .schema.SchemaRegistry import SchemaRegistry, schema_registry
from .schema.endpoint_schema import EndpointSchema
from .schema.schema_detector import SchemaDetector, SchemaField
from .schema.schema_registry import SchemaRegistry, schema_registry
__all__ = [
"EndpointMonitor",

View file

@ -18,10 +18,10 @@ from typing import Optional
import httpx
from .endpoint_definition import EndpointDefinition
from .endpoint_monitoring_result import MonitoringResult
from ..schema.schema_registry import SchemaRegistry, schema_registry
from ...config import settings
from ..schema.SchemaRegistry import SchemaRegistry, schema_registry
from .EndpointDefinition import EndpointDefinition
from .EndpointMonitoringResult import MonitoringResult
logger = logging.getLogger(__name__)

View file

@ -13,7 +13,7 @@ from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any, Optional
from .SchemaField import SchemaField
from .schema_field import SchemaField
logger = logging.getLogger(__name__)

View file

@ -13,7 +13,7 @@ import logging
from datetime import datetime
from typing import Any
from .SchemaField import SchemaField
from .schema_field import SchemaField
logger = logging.getLogger(__name__)

View file

@ -16,9 +16,9 @@ from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Optional
from .endpoint_schema import EndpointSchema
from .schema_detector import SchemaDetector
from ...config import settings
from .EndpointSchema import EndpointSchema
from .SchemaDetector import SchemaDetector
logger = logging.getLogger(__name__)

View file

@ -11,9 +11,9 @@ from typing import Optional
from .game_service import GameService
from .profile_service import ProfileService
from ..api.client import GeoGuessrClient
from ..models.Game import Game
from ..monitoring.schema.SchemaRegistry import schema_registry
from ..api import GeoGuessrClient
from ..models import Game
from ..monitoring import schema_registry
logger = logging.getLogger(__name__)

View file

@ -7,11 +7,8 @@ Handles game history, details, and competitive data with dynamic schema support.
import logging
from typing import Optional
from ..api.client import DynamicResponse, GeoGuessrClient
from ..api.endpoints import Endpoints
from ..models.DailyChallenge import DailyChallenge
from ..models.Game import Game
from ..models.SeasonStats import SeasonStats
from ..api import Endpoints, DynamicResponse, GeoGuessrClient
from ..models import DailyChallenge, Game, SeasonStats
logger = logging.getLogger(__name__)

View file

@ -8,11 +8,8 @@ dynamic schema adaptation.
import logging
from typing import Optional
from ..api.client import DynamicResponse, GeoGuessrClient
from ..api.endpoints import Endpoints
from ..models.Achievement import Achievement
from ..models.UserProfile import UserProfile
from ..models.UserStats import UserStats
from ..api import DynamicResponse, GeoGuessrClient, Endpoints
from ..models import Achievement, UserProfile, UserStats
logger = logging.getLogger(__name__)