Coverage for src/lexigram/graphql/types.py: 99%
187 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 04:37 +0800
1"""Core types and enums for lexigram-graphql.
3This module defines all type definitions, enums, and DTOs
4used throughout the GraphQL package.
6Following the Root File Pattern for easy access:
7 from lexigram.graphql.types import OperationType, GraphQLErrorCode
8"""
10from __future__ import annotations
12from collections.abc import Callable
13from dataclasses import dataclass, field
14from datetime import UTC, datetime
15from enum import StrEnum
16from typing import Any, TypeVar
17from uuid import uuid4
19from lexigram.contracts.core import ClockProtocol as _Clock
20from lexigram.contracts.core import IdGeneratorProtocol as _Identity
23def _now(clock: _Clock | None = None) -> datetime:
24 """Get current timestamp using injected clock."""
25 if clock is not None:
26 return clock.now()
27 return datetime.now(UTC)
30def _uuid(identity: _Identity | None = None) -> str:
31 """Generate a UUID using injected identity."""
32 if identity is not None:
33 return identity.generate()
34 return str(uuid4())
37# =============================================================================
38# Enums
39# =============================================================================
42class OperationType(StrEnum):
43 """GraphQL operation types."""
45 QUERY = "query"
46 MUTATION = "mutation"
47 SUBSCRIPTION = "subscription"
50class GraphQLErrorCode(StrEnum):
51 """Standard GraphQL error codes."""
53 # Client errors
54 VALIDATION_ERROR = "VALIDATION_ERROR"
55 BAD_USER_INPUT = "BAD_USER_INPUT"
56 UNAUTHENTICATED = "UNAUTHENTICATED"
57 UNAUTHORIZED = "UNAUTHORIZED"
58 FORBIDDEN = "FORBIDDEN"
59 NOT_FOUND = "NOT_FOUND"
61 # Server errors
62 INTERNAL_SERVER_ERROR = "INTERNAL_SERVER_ERROR"
63 SERVICE_UNAVAILABLE = "SERVICE_UNAVAILABLE"
65 # GraphQL specific
66 GRAPHQL_PARSE_FAILED = "GRAPHQL_PARSE_FAILED"
67 GRAPHQL_VALIDATION_FAILED = "GRAPHQL_VALIDATION_FAILED"
68 PERSISTED_QUERY_NOT_FOUND = "PERSISTED_QUERY_NOT_FOUND"
70 # Rate limiting
71 RATE_LIMITED = "RATE_LIMITED"
72 QUERY_TOO_COMPLEX = "QUERY_TOO_COMPLEX"
73 QUERY_TOO_DEEP = "QUERY_TOO_DEEP"
76class DirectiveLocation(StrEnum):
77 """GraphQL directive locations."""
79 # Executable locations
80 QUERY = "QUERY"
81 MUTATION = "MUTATION"
82 SUBSCRIPTION = "SUBSCRIPTION"
83 FIELD = "FIELD"
84 FRAGMENT_DEFINITION = "FRAGMENT_DEFINITION"
85 FRAGMENT_SPREAD = "FRAGMENT_SPREAD"
86 INLINE_FRAGMENT = "INLINE_FRAGMENT"
87 VARIABLE_DEFINITION = "VARIABLE_DEFINITION"
89 # Type system locations
90 SCHEMA = "SCHEMA"
91 SCALAR = "SCALAR"
92 OBJECT = "OBJECT"
93 FIELD_DEFINITION = "FIELD_DEFINITION"
94 ARGUMENT_DEFINITION = "ARGUMENT_DEFINITION"
95 INTERFACE = "INTERFACE"
96 UNION = "UNION"
97 ENUM = "ENUM"
98 ENUM_VALUE = "ENUM_VALUE"
99 INPUT_OBJECT = "INPUT_OBJECT"
100 INPUT_FIELD_DEFINITION = "INPUT_FIELD_DEFINITION"
103class CacheScope(StrEnum):
104 """Cache control scope for responses."""
106 PUBLIC = "PUBLIC"
107 PRIVATE = "PRIVATE"
110class SubscriptionProtocol(StrEnum):
111 """WebSocket subscription protocols."""
113 GRAPHQL_WS = "graphql-ws"
114 GRAPHQL_TRANSPORT_WS = "graphql-transport-ws"
117# =============================================================================
118# Data Transfer Objects
119# =============================================================================
122@dataclass
123class GraphQLLocation:
124 """Location in a GraphQL document."""
126 line: int
127 column: int
130@dataclass
131class GraphQLErrorExtensions:
132 """Extended error information."""
134 code: str
135 timestamp: datetime = field(default_factory=_now)
136 request_id: str | None = None
137 path: list[str] | None = None
138 field: str | None = None
139 argument: str | None = None
140 exception: str | None = None
141 stacktrace: list[str] | None = None
143 def to_dict(self) -> dict[str, Any]:
144 """Convert to dictionary for JSON serialization."""
145 result: dict[str, Any] = {"code": self.code}
146 if self.timestamp:
147 result["timestamp"] = self.timestamp.isoformat()
148 if self.request_id:
149 result["requestId"] = self.request_id
150 if self.path:
151 result["path"] = self.path
152 if self.field:
153 result["field"] = self.field
154 if self.argument:
155 result["argument"] = self.argument
156 # Only include exception/stacktrace in debug mode
157 return result
160@dataclass
161class GraphQLErrorData:
162 """Structured GraphQL error data."""
164 message: str
165 locations: list[GraphQLLocation] | None = None
166 path: list[str | int] | None = None
167 extensions: GraphQLErrorExtensions | None = None
169 def to_dict(self) -> dict[str, Any]:
170 """Convert to standard GraphQL error format."""
171 result: dict[str, Any] = {"message": self.message}
173 if self.locations:
174 result["locations"] = [
175 {"line": loc.line, "column": loc.column} for loc in self.locations
176 ]
178 if self.path:
179 result["path"] = self.path
181 if self.extensions:
182 result["extensions"] = self.extensions.to_dict()
184 return result
187@dataclass
188class FieldInfo:
189 """Information about a GraphQL field."""
191 name: str
192 parent_type: str
193 return_type: str
194 arguments: dict[str, Any] = field(default_factory=dict)
195 directives: list[str] = field(default_factory=list)
196 is_nullable: bool = True
197 is_list: bool = False
198 deprecation_reason: str | None = None
201@dataclass
202class OperationInfo:
203 """Information about a GraphQL operation."""
205 name: str | None
206 operation_type: OperationType
207 variables: dict[str, Any] = field(default_factory=dict)
208 selection_count: int = 0
209 depth: int = 0
210 complexity: int = 0
213@dataclass
214class ResolverInfo:
215 """Context information passed to resolvers."""
217 field_name: str
218 parent_type: str
219 return_type: str
220 path: list[str | int]
221 operation: OperationInfo
222 variables: dict[str, Any]
223 context: Any
224 root_value: Any = None
227@dataclass
228class CacheControl:
229 """Cache control settings for a field or type."""
231 max_age: int = 0
232 scope: CacheScope = CacheScope.PUBLIC
233 inherit_max_age: bool = False
235 def to_header(self) -> str:
236 """Generate Cache-Control header value."""
237 parts = []
238 if self.scope == CacheScope.PRIVATE:
239 parts.append("private")
240 else:
241 parts.append("public")
242 parts.append(f"max-age={self.max_age}")
243 return ", ".join(parts)
246@dataclass
247class QueryMetrics:
248 """Metrics for a GraphQL query execution."""
250 request_id: str = field(default_factory=_uuid)
251 operation_name: str | None = None
252 operation_type: OperationType | None = None
253 started_at: datetime = field(default_factory=_now)
254 ended_at: datetime | None = None
255 duration_ms: float = 0.0
256 field_count: int = 0
257 depth: int = 0
258 complexity: int = 0
259 errors_count: int = 0
260 cache_hit: bool = False
262 def complete(self) -> None:
263 """Mark the query as complete and calculate duration."""
264 self.ended_at = _now()
265 self.duration_ms = (self.ended_at - self.started_at).total_seconds() * 1000
268@dataclass
269class SubscriptionInfo:
270 """Information about an active subscription."""
272 subscription_id: str
273 operation_name: str | None
274 query: str
275 variables: dict[str, Any]
276 created_at: datetime = field(default_factory=_now)
277 last_event_at: datetime | None = None
278 event_count: int = 0
281@dataclass
282class DataLoaderStats:
283 """Statistics for a DataLoaderProtocol instance."""
285 name: str
286 batch_count: int = 0
287 load_count: int = 0
288 cache_hits: int = 0
289 cache_misses: int = 0
291 @property
292 def cache_hit_ratio(self) -> float:
293 """Calculate cache hit ratio."""
294 total = self.cache_hits + self.cache_misses
295 return self.cache_hits / total if total > 0 else 0.0
298# =============================================================================
299# Type Variables
300# =============================================================================
302T = TypeVar("T")
303K = TypeVar("K")
304V = TypeVar("V")
306# Handler types
307ResolverFunc = Callable[..., Any]
308MiddlewareFunc = Callable[[ResolverInfo, Callable], Any]
309ErrorHandler = Callable[[Exception, ResolverInfo], GraphQLErrorData]
312__all__ = [
313 "CacheControl",
314 "CacheScope",
315 "DataLoaderStats",
316 "DirectiveLocation",
317 "ErrorHandler",
318 "FieldInfo",
319 "GraphQLErrorCode",
320 "GraphQLErrorData",
321 "GraphQLErrorExtensions",
322 # DTOs
323 "GraphQLLocation",
324 "K",
325 "MiddlewareFunc",
326 "OperationInfo",
327 # Enums
328 "OperationType",
329 "QueryMetrics",
330 "ResolverFunc",
331 "ResolverInfo",
332 "SubscriptionInfo",
333 "SubscriptionProtocol",
334 # Type vars
335 "T",
336 "V",
337]