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 UTC, datetime 

10from typing import Generic, TypeVar 

11 

12from pydantic import BaseModel, Field, field_validator, model_validator 

13 

14T = TypeVar("T") 

15 

16 

17# ============================================================================ 

18# Meta & pagination 

19# ============================================================================ 

20 

21 

22class APIResponseMeta(BaseModel): 

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

24 

25 timestamp: str = Field( 

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

27 description="ISO 8601 response timestamp", 

28 ) 

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

30 request_id: str = Field(default="", description="Unique request ID for tracing") 

31 

32 

33class PaginationMeta(BaseModel): 

34 """Cursor-based pagination metadata.""" 

35 

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

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

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

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

40 next_cursor: str | None = Field(default=None, description="Opaque cursor for next page") 

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

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

43 

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

45 @classmethod 

46 def compute_total_pages(cls, v, info): 

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

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

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

50 return v 

51 

52 @model_validator(mode="after") 

53 def compute_pagination_flags(self): 

54 self.has_next = self.page < self.total_pages 

55 self.has_prev = self.page > 1 

56 return self 

57 

58 

59# ============================================================================ 

60# Error detail (RFC 9457 Problem Details) 

61# ============================================================================ 

62 

63 

64class APIErrorDetail(BaseModel): 

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

66 

67 type: str = Field( 

68 default="about:blank", 

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

70 ) 

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

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

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

74 instance: str | None = Field( 

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

76 ) 

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

78 field: str | None = Field(default=None, description="Field name for validation errors") 

79 

80 

81# ============================================================================ 

82# Response envelope 

83# ============================================================================ 

84 

85 

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

87 """Canonical API response envelope. 

88 

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

90 

91 Success: 

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

93 

94 Error: 

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

96 """ 

97 

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

99 data: T | None = Field(default=None, description="Response payload") 

100 error: APIErrorDetail | None = Field( 

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

102 ) 

103 meta: APIResponseMeta = Field(default_factory=APIResponseMeta, description="Response metadata") 

104 

105 

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

107 """Paginated list response.""" 

108 

109 success: bool = Field(default=True) 

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

111 pagination: PaginationMeta = Field( 

112 default_factory=PaginationMeta, description="Pagination metadata" 

113 ) 

114 meta: APIResponseMeta = Field(default_factory=APIResponseMeta, description="Response metadata") 

115 

116 

117# ============================================================================ 

118# Health & version 

119# ============================================================================ 

120 

121 

122class HealthComponent(BaseModel): 

123 """Individual component health status.""" 

124 

125 name: str 

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

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

128 error: str | None = Field(default=None) 

129 

130 

131class HealthResponse(BaseModel): 

132 """Full health check response.""" 

133 

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

135 uptime_seconds: float 

136 components: list[HealthComponent] = Field(default_factory=list) 

137 timestamp: str = Field(default_factory=lambda: datetime.now(UTC).isoformat()) 

138 

139 

140class VersionResponse(BaseModel): 

141 """Version info response.""" 

142 

143 version: str 

144 build: str = "" 

145 commit_sha: str | None = None 

146 python_version: str = "" 

147 environment: str = "production"