Coverage for functions \ flipdare \ request \ request_adapter.py: 95%

1#!/usr/bin/env python 

2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved. 

3# 

4# This file is part of Flipdare's proprietary software and contains 

5# confidential and copyrighted material. Unauthorised copying, 

6# modification, distribution, or use of this file is strictly 

7# prohibited without prior written permission from Flipdare Pty Ltd. 

8# 

9# This software includes third-party components licensed under MIT, 

10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details. 

11# 

12 

13from abc import ABC 

14from dataclasses import dataclass, field, replace 

15from typing import Any, ClassVar, Self, cast, override 

16import flask 

17from firebase_functions import https_fn 

18from pydantic import TypeAdapter, ValidationError 

19 

20from flipdare.app_log import LOG 

21from flipdare.app_types import JsonDict, SchemaDict 

22from flipdare.constants import IS_DEBUG, IS_TRACE 

23from flipdare.core.proto_unwrapper import ProtoUnwrapper 

24from flipdare.error.app_error import AppError 

25from flipdare.error.data_load_error import DataLoadError 

26from flipdare.error.message_format import ValidationErrorMsgFormat 

27 

28from flipdare.request.app_request import AppRequest, AppRequestType 

29from flipdare.request.request_types import AppHttpRequest, AppHttpRequestType 

30from flipdare.request.request_validator import RequestValidator 

31 

32__all__ = ["RequestAdapter", "RequestAdapterError"] 

33 

34 

35@dataclass(frozen=True) 

36class RequestAdapterError: 

37 endpoint: str 

38 class_name: type[Any] 

39 errors: list[str] = field(default_factory=list) 

40 parse_failed: bool = False 

41 

42 def add(self, error: str, parse_failed: bool = False) -> Self: 

43 return replace(self, errors=[*self.errors, error], parse_failed=parse_failed) 

44 

45 @property 

46 def user_error_code(self) -> str: 

47 from flipdare.message.user_error_code import UserErrorCode 

48 

49 ct = len(self.errors) + 1 

50 return UserErrorCode.validation(self.class_name, ct, parse_failed=self.parse_failed) 

51 

52 def error(self) -> AppError: 

53 formatter = ValidationErrorMsgFormat( 

54 class_type=self.class_name, 

55 error=self.errors, 

56 parse_failed=self.parse_failed, 

57 ) 

58 return DataLoadError.malformed( 

59 endpoint=self.endpoint, 

60 missing_code=formatter.user_error_code, 

61 error=self.errors, 

62 ) 

63 

64 

65class RequestAdapter[TSchema: SchemaDict](ABC): 

66 _request: AppRequest[AppRequestType] 

67 _authenticated_uid: str | None 

68 _endpoint: str 

69 _params: JsonDict 

70 _result: RequestAdapterError | TSchema 

71 

72 SCHEMA_CLS: type[TSchema] 

73 VALIDATORS: ClassVar[tuple[type[RequestValidator], ...]] = () 

74 

75 def __init__(self, request: AppRequest[AppRequestType]) -> None: 

76 endpoint, params = RequestAdapter.request_parts(request.request_type) 

77 # Only set authenticated_uid if auth was already attempted 

78 self._authenticated_uid = ( 

79 request.auth_result.user_id if request.auth_result is not None else None 

80 ) 

81 self._request = request 

82 self._endpoint = endpoint 

83 self._params = params 

84 self._validate() 

85 

86 @classmethod 

87 def from_callable(cls, req: https_fn.CallableRequest[Any]) -> Self: 

88 request = AppRequest.callable(req) 

89 return cls(request) 

90 

91 @classmethod 

92 def from_http(cls, req: flask.Request, req_type: AppHttpRequestType) -> Self: 

93 request = AppRequest.http(req, req_type) 

94 return cls(request) 

95 

96 @staticmethod 

97 def request_parts( 

98 req: https_fn.CallableRequest[JsonDict] | AppHttpRequest, 

99 ) -> tuple[str, dict[str, Any]]: 

100 if isinstance(req, https_fn.CallableRequest): 

101 # Callable requests carry structured data directly; URL is for debugging only. 

102 params = req.data 

103 # Use .endpoint from the underlying raw request 

104 raw = req.raw_request 

105 endpoint = raw.endpoint or raw.url or "unknown" 

106 

107 if IS_DEBUG: 

108 debug_str = "\n\t".join(f"{k}: {v}" for k, v in params.items()) 

109 msg = f"CallableRequest: endpoint={endpoint}\n\t{debug_str}" 

110 LOG().debug(msg) 

111 

112 return endpoint, params 

113 # AppHttpRequest wraps the raw flask.Request — unwrap it first. 

114 raw = req.raw_request 

115 endpoint = raw.endpoint or raw.url or "unknown" 

116 if raw.method.upper() == "POST": 

117 if raw.is_json: 

118 params = cast("dict[str, Any]", raw.get_json(silent=True) or {}) 

119 else: 

120 # Fallback to form data, converted to a real dict 

121 params = raw.form.to_dict() 

122 else: 

123 # For GET, raw.args is already parsed. to_dict() makes it a standard dict. 

124 params = raw.args.to_dict() 

125 

126 if IS_DEBUG: 

127 debug_str = "\n\t".join(f"{k}: {v}" for k, v in params.items()) 

128 msg = f"AppHttpRequest: endpoint={endpoint}, method={raw.method}\n\t{debug_str}" 

129 LOG().debug(msg) 

130 

131 return endpoint, params 

132 

133 @property 

134 def request(self) -> AppRequest[AppRequestType]: 

135 return self._request 

136 

137 @property 

138 def endpoint(self) -> str: 

139 return self._endpoint 

140 

141 @property 

142 def ip_address(self) -> str: 

143 return self._request.ip_address 

144 

145 @property 

146 def authenticated_uid(self) -> str | None: 

147 return self._authenticated_uid 

148 

149 @property 

150 def data(self) -> TSchema: 

151 result = self._result 

152 if isinstance(result, RequestAdapterError): 

153 raise result.error() 

154 

155 return result 

156 

157 @property 

158 def params(self) -> JsonDict: 

159 return self._params 

160 

161 @property 

162 def user_error_code(self) -> str | None: 

163 if isinstance(self._result, RequestAdapterError): 

164 return self._result.user_error_code 

165 return None 

166 

167 def validation_errors(self) -> list[str] | None: 

168 if isinstance(self._result, RequestAdapterError): 

169 return self._result.errors 

170 return None 

171 

172 def validate(self) -> None: 

173 result = self._result 

174 if isinstance(result, RequestAdapterError): 

175 raise result.error() 

176 

177 def _validate(self) -> None: 

178 parsed = self._parse() 

179 errors: list[str] = [] 

180 

181 if isinstance(parsed, RequestAdapterError): 

182 # since we dont have parsed , we cant run validators .. 

183 self._result = parsed 

184 return 

185 

186 for validator_cls in self.VALIDATORS: 

187 errors.extend(validator_cls(cast("JsonDict", parsed)).validate()) 

188 

189 if errors: 

190 self._result = RequestAdapterError( 

191 endpoint=self._endpoint, 

192 class_name=self.SCHEMA_CLS, 

193 errors=errors, 

194 parse_failed=False, 

195 ) 

196 else: 

197 self._result = parsed 

198 

199 def _parse(self) -> RequestAdapterError | TSchema: 

200 params = self._params 

201 try: 

202 json = ProtoUnwrapper(params).unwrap() 

203 adapter = TypeAdapter(self.SCHEMA_CLS) 

204 data = adapter.validate_python(json) 

205 if IS_TRACE: 

206 msg = f"Successfully parsed request data for endpoint {self.endpoint}: {data}" 

207 LOG().trace(msg) 

208 return data 

209 except ValidationError as e: 

210 errors = DataLoadError.parse_error(e) 

211 LOG().error(f"Error parsing request data {self.endpoint}: {e}\nERRORS={errors}") 

212 return RequestAdapterError( 

213 endpoint=self.endpoint, 

214 class_name=self.SCHEMA_CLS, 

215 errors=errors, 

216 parse_failed=True, 

217 ) 

218 

219 @override 

220 def __repr__(self) -> str: 

221 return ( 

222 f"{self.__class__.__name__}(endpoint={self.endpoint}, " 

223 f"authenticated_uid={self.authenticated_uid}, params={self.params})" 

224 ) 

225 

226 @override 

227 def __str__(self) -> str: 

228 return self.__repr__()