Coverage for agentos/models/response.py: 89%

1"""AgentOS Response Models — RFC 9457 compatible, OpenAPI-ready. 

2 

3Provides the canonical API response envelope used across all endpoints. 

4All responses are wrapped in APIResponse[T] for consistency. 

5""" 

6 

7from __future__ import annotations 

8 

9from datetime import datetime, timezone 

10from typing import Any, Generic, List, Optional, TypeVar 

11 

12from pydantic import BaseModel, Field, field_validator, model_validator 

13 

14 

15T = TypeVar("T") 

16 

17 

18# ============================================================================ 

19# Meta & pagination 

20# ============================================================================ 

21 

22class APIResponseMeta(BaseModel): 

23 """Response metadata: timing, version, request tracking.""" 

24 

25 timestamp: str = Field( 

26 default_factory=lambda: datetime.now(timezone.utc).isoformat(), 

27 description="ISO 8601 response timestamp", 

28 ) 

29 version: str = Field(default="1.0", description="API version") 

30 request_id: str = Field( 

31 default="", description="Unique request ID for tracing" 

32 ) 

33 

34 

35class PaginationMeta(BaseModel): 

36 """Cursor-based pagination metadata.""" 

37 

38 page: int = Field(default=1, ge=1, description="Current page (1-based)") 

39 page_size: int = Field(default=20, ge=1, le=200, description="Items per page") 

40 total_items: int = Field(default=0, ge=0, description="Total matching items") 

41 total_pages: int = Field(default=0, ge=0, description="Total pages") 

42 next_cursor: Optional[str] = Field( 

43 default=None, description="Opaque cursor for next page" 

44 ) 

45 has_next: bool = Field(default=False, description="Whether a next page exists") 

46 has_prev: bool = Field(default=False, description="Whether a previous page exists") 

47 

48 @field_validator("total_pages", mode="before") 

49 @classmethod 

50 def compute_total_pages(cls, v, info): 

51 if v == 0 and info.data.get("total_items", 0) > 0: 

52 page_size = info.data.get("page_size", 20) 

53 return max(1, (info.data["total_items"] + page_size - 1) // page_size) 

54 return v 

55 

56 @model_validator(mode="after") 

57 def compute_pagination_flags(self): 

58 self.has_next = self.page < self.total_pages 

59 self.has_prev = self.page > 1 

60 return self 

61 

62 

63# ============================================================================ 

64# Error detail (RFC 9457 Problem Details) 

65# ============================================================================ 

66 

67class APIErrorDetail(BaseModel): 

68 """Single error entry conforming to RFC 9457 Problem Details.""" 

69 

70 type: str = Field( 

71 default="about:blank", 

72 description="URI reference identifying the problem type", 

73 ) 

74 title: str = Field(description="Short, human-readable summary") 

75 status: int = Field(default=500, ge=100, le=599) 

76 detail: str = Field(default="", description="Human-readable explanation") 

77 instance: Optional[str] = Field( 

78 default=None, description="URI reference identifying the specific occurrence" 

79 ) 

80 code: str = Field(default="INTERNAL_ERROR", description="Machine-readable error code") 

81 field: Optional[str] = Field( 

82 default=None, description="Field name for validation errors" 

83 ) 

84 

85 

86# ============================================================================ 

87# Response envelope 

88# ============================================================================ 

89 

90class APIResponse(BaseModel, Generic[T]): 

91 """Canonical API response envelope. 

92 

93 All endpoints return this structure with the generic type T for the data field. 

94 

95 Success: 

96 {"success": true, "data": {...}, "meta": {...}} 

97 

98 Error: 

99 {"success": false, "error": {...}, "meta": {...}} 

100 """ 

101 

102 success: bool = Field(default=True, description="Whether the request succeeded") 

103 data: Optional[T] = Field(default=None, description="Response payload") 

104 error: Optional[APIErrorDetail] = Field( 

105 default=None, description="Error detail (only when success=False)" 

106 ) 

107 meta: APIResponseMeta = Field( 

108 default_factory=APIResponseMeta, description="Response metadata" 

109 ) 

110 

111 

112class PaginatedResponse(BaseModel, Generic[T]): 

113 """Paginated list response.""" 

114 

115 success: bool = Field(default=True) 

116 data: List[T] = Field(default_factory=list, description="Page items") 

117 pagination: PaginationMeta = Field( 

118 default_factory=PaginationMeta, description="Pagination metadata" 

119 ) 

120 meta: APIResponseMeta = Field( 

121 default_factory=APIResponseMeta, description="Response metadata" 

122 ) 

123 

124 

125# ============================================================================ 

126# Health & version 

127# ============================================================================ 

128 

129class HealthComponent(BaseModel): 

130 """Individual component health status.""" 

131 

132 name: str 

133 status: str = Field(description="healthy | degraded | unhealthy") 

134 latency_ms: float = Field(default=0.0, description="Check latency in ms") 

135 error: Optional[str] = Field(default=None) 

136 

137 

138class HealthResponse(BaseModel): 

139 """Full health check response.""" 

140 

141 status: str = Field(description="healthy | degraded | unhealthy") 

142 uptime_seconds: float 

143 components: List[HealthComponent] = Field(default_factory=list) 

144 timestamp: str = Field( 

145 default_factory=lambda: datetime.now(timezone.utc).isoformat() 

146 ) 

147 

148 

149class VersionResponse(BaseModel): 

150 """Version info response.""" 

151 

152 version: str 

153 build: str = "" 

154 commit_sha: Optional[str] = None 

155 python_version: str = "" 

156 environment: str = "production"