Coverage for fastapi_restly / views / _openapi.py: 92%

1""" 

2Internal OpenAPI post-processing: x-resource-ref annotations, and a warning 

3when two classes in the spec have the same name (see ``_openapi_names``). 

4 

5Called automatically by include_view() — no public API. ``fr.configure(app)`` 

6and ``configure_tests(app=app)`` also patch the app, so the name check runs 

7when all views are on an APIRouter. 

8 

9FK columns and SQLAlchemy relationship fields backed by IDSchema/IDRef 

10are annotated with ``x-resource-ref: "<resource-name>"`` in the generated spec. 

11Full nested-object relationships (plain Pydantic model fields) are left untouched. 

12""" 

13 

14import inspect 

15import types 

16import weakref 

17from dataclasses import dataclass 

18from typing import Any, Union, get_args, get_origin 

19 

20import fastapi 

21import pydantic 

22from sqlalchemy import inspect as sa_inspect 

23 

24from .._mapping import is_mapped_class 

25from ..schemas import IDSchema 

26from ._openapi_names import _warn_on_name_clashes 

27 

28_PATCHED_ATTR = "_fr_resource_refs_patched" 

29 

30 

31@dataclass 

32class _Entry: 

33 model: type[Any] 

34 resource_name: str 

35 schema: type[pydantic.BaseModel] 

36 schema_response: type[pydantic.BaseModel] 

37 schema_create: type[pydantic.BaseModel] 

38 schema_update: type[pydantic.BaseModel] 

39 

40 

41_RegistryKey = int 

42_RegistryRef = weakref.ReferenceType[fastapi.FastAPI | fastapi.APIRouter] 

43_registry: dict[_RegistryKey, tuple[_RegistryRef, list[_Entry]]] = {} 

44 

45 

46def _registry_entries( 

47 parent_router: fastapi.FastAPI | fastapi.APIRouter, 

48) -> list[_Entry]: 

49 """Return resource-ref entries for ``parent_router``. 

50 

51 ``APIRouter`` instances are weak-referenceable but not hashable, so 

52 ``WeakKeyDictionary`` cannot store them. Key by identity and remove the 

53 entry when the router/app is collected. 

54 """ 

55 key = id(parent_router) 

56 existing = _registry.get(key) 

57 if existing is not None: 

58 ref, entries = existing 

59 if ref() is parent_router: 59 ↛ 62line 59 didn't jump to line 62 because the condition on line 59 was always true

60 return entries 

61 

62 def cleanup(ref: _RegistryRef, key: _RegistryKey = key) -> None: 

63 current = _registry.get(key) 

64 if current is not None and current[0] is ref: 64 ↛ exitline 64 didn't return from function 'cleanup' because the condition on line 64 was always true

65 _registry.pop(key, None) 

66 

67 ref = weakref.ref(parent_router, cleanup) 

68 entries: list[_Entry] = [] 

69 _registry[key] = (ref, entries) 

70 return entries 

71 

72 

73def _registered_entries( 

74 parent_router: fastapi.FastAPI | fastapi.APIRouter, 

75) -> list[_Entry]: 

76 existing = _registry.get(id(parent_router)) 

77 if existing is None: 

78 return [] 

79 ref, entries = existing 

80 if ref() is parent_router: 80 ↛ 82line 80 didn't jump to line 82 because the condition on line 80 was always true

81 return entries 

82 return [] 

83 

84 

85def _register_for_resource_ref( 

86 parent_router: fastapi.FastAPI | fastapi.APIRouter, view_cls: type 

87) -> None: 

88 """Register a view's model→resource mapping and ensure the spec is patched. 

89 

90 Silently skips views without a SQLAlchemy model (e.g. plain View subclasses). 

91 """ 

92 model = getattr(view_cls, "model", None) 

93 if not is_mapped_class(model): 

94 return 

95 

96 resource_name = "".join( 

97 c.__dict__["prefix"] for c in reversed(view_cls.mro()) if "prefix" in c.__dict__ 

98 ).lstrip("/") 

99 

100 entry = _Entry( 

101 model=model, 

102 resource_name=resource_name, 

103 schema=view_cls.schema, 

104 schema_response=view_cls.schema_response, 

105 schema_create=view_cls.schema_create, 

106 schema_update=view_cls.schema_update, 

107 ) 

108 

109 entries = _registry_entries(parent_router) 

110 entries.append(entry) 

111 

112 # Only the FastAPI app generates the OpenAPI spec; APIRouter parents have 

113 # no ``.openapi`` to patch. Views registered on a router lose x-resource-ref 

114 # annotations until the framework can walk to a root app, which is a 

115 # separate gap. The name check reads the app's routes, so it covers them 

116 # when fr.configure(app) or configure_tests(app=app) patched the app. 

117 if isinstance(parent_router, fastapi.FastAPI): 

118 _ensure_patched(parent_router) 

119 

120 

121def _ensure_patched(app: fastapi.FastAPI) -> None: 

122 """Wrap app.openapi() once so annotations are injected on first call, and 

123 the spec is checked for two classes with the same name.""" 

124 if getattr(app, _PATCHED_ATTR, False): 

125 return 

126 

127 original_openapi = app.openapi 

128 

129 def patched_openapi() -> dict[str, Any]: 

130 spec = original_openapi() 

131 entries = _registered_entries(app) 

132 model_to_resource = {e.model: e.resource_name for e in entries} 

133 _annotate_spec(spec, entries, model_to_resource) 

134 _warn_on_name_clashes(app, spec) 

135 return spec 

136 

137 app.openapi = patched_openapi # type: ignore[method-assign] 

138 setattr(app, _PATCHED_ATTR, True) 

139 

140 

141def _is_id_ref_annotation(annotation: Any) -> bool: 

142 """Return True if annotation is IDSchema[X], IDRef[X], or list/Optional thereof. 

143 

144 Returns False for full nested Pydantic model objects — those are not ID references. 

145 Concrete user-defined subclasses like ``AuthorSchema(IDSchema)`` return False; 

146 only parametrized generics like ``IDSchema[Author]`` or ``IDRef[Author]`` 

147 return True, since those represent model ID references. 

148 """ 

149 origin = get_origin(annotation) 

150 

151 # Unwrap Optional / Union (X | None, Optional[X], Union[X, Y]) 

152 if origin in (Union, types.UnionType): 

153 return any( 

154 _is_id_ref_annotation(a) 

155 for a in get_args(annotation) 

156 if a is not type(None) 

157 ) 

158 

159 # list[X] — check the element type 

160 if origin is list: 

161 args = get_args(annotation) 

162 return bool(args and _is_id_ref_annotation(args[0])) 

163 

164 # Check for parametrized IDSchema/IDRef generics. 

165 # In Pydantic v2, IDSchema[Author] and IDRef[Author] are concrete classes, 

166 # so inspect.isclass() returns True for them too. We distinguish via 

167 # __pydantic_generic_metadata__["origin"]: 

168 # - Parametrized: IDSchema[Author] → origin = IDSchema 

169 # - Parametrized: IDRef[Author] → origin = IDRef 

170 # - User-defined subclass: AuthorSchema(IDSchema) → origin = None (not a parametrization) 

171 pydantic_meta = getattr(annotation, "__pydantic_generic_metadata__", {}) 

172 origin_cls = pydantic_meta.get("origin") 

173 if inspect.isclass(origin_cls): 

174 try: 

175 return issubclass(origin_cls, IDSchema) 

176 except TypeError: 

177 return False 

178 

179 return False 

180 

181 

182def _field_openapi_key(schema_cls: type[pydantic.BaseModel], field_name: str) -> str: 

183 """Return the OpenAPI property key for a field, respecting serialization aliases.""" 

184 field_info = schema_cls.model_fields.get(field_name) 

185 if field_info is None: 185 ↛ 186line 185 didn't jump to line 186 because the condition on line 185 was never true

186 return field_name 

187 if field_info.serialization_alias: 187 ↛ 188line 187 didn't jump to line 188 because the condition on line 187 was never true

188 return field_info.serialization_alias 

189 if field_info.alias: 189 ↛ 190line 189 didn't jump to line 190 because the condition on line 189 was never true

190 return field_info.alias 

191 return field_name 

192 

193 

194def _compute_refs( 

195 schema_cls: type[pydantic.BaseModel], 

196 model_cls: type[Any], 

197 model_to_resource: dict[type[Any], str], 

198) -> dict[str, str]: 

199 """Return {openapi_property_key: resource_name} for FK columns and ID-ref relationship fields.""" 

200 result: dict[str, str] = {} 

201 try: 

202 mapper = sa_inspect(model_cls) 

203 except Exception: 

204 return result 

205 

206 for field_name, field_info in schema_cls.model_fields.items(): 

207 resource_name: str | None = None 

208 

209 if field_name in mapper.columns: 

210 fks = list(mapper.columns[field_name].foreign_keys) 

211 if fks: 

212 target_table = fks[0].column.table # Table object identity 

213 for m in mapper.registry.mappers: 213 ↛ 224line 213 didn't jump to line 224 because the loop on line 213 didn't complete

214 if m.local_table is target_table: 

215 resource_name = model_to_resource.get(m.class_) 

216 break 

217 

218 elif field_name in mapper.relationships: 

219 # Only annotate if the schema field carries ID references, not full nested objects. 

220 if _is_id_ref_annotation(field_info.annotation): 

221 target_model = mapper.relationships[field_name].mapper.class_ 

222 resource_name = model_to_resource.get(target_model) 

223 

224 if resource_name is not None: 

225 result[_field_openapi_key(schema_cls, field_name)] = resource_name 

226 

227 return result 

228 

229 

230def _annotate_spec( 

231 spec: dict[str, Any], entries: list[_Entry], model_to_resource: dict[type[Any], str] 

232) -> None: 

233 """Mutate spec in-place, adding x-resource-ref to qualifying properties.""" 

234 schemas = spec.get("components", {}).get("schemas", {}) 

235 if not schemas: 

236 return 

237 

238 for entry in entries: 

239 # Each class gets the refs of its own fields: a class that the view 

240 # sets can give a field another type, such as a nested object for an 

241 # id. The view's schema is in the spec only when another schema nests 

242 # it or a custom route names it. 

243 for schema_cls in dict.fromkeys( 

244 ( 

245 entry.schema, 

246 entry.schema_response, 

247 entry.schema_create, 

248 entry.schema_update, 

249 ) 

250 ): 

251 refs = _compute_refs(schema_cls, entry.model, model_to_resource) 

252 props = schemas.get(schema_cls.__name__, {}).get("properties", {}) 

253 for prop_key, resource_name in refs.items(): 

254 if prop_key in props: 

255 props[prop_key]["x-resource-ref"] = resource_name