|
| 1 | +import datetime |
| 2 | +import hashlib |
| 3 | +import json |
| 4 | +import re |
| 5 | +from typing import Any, Dict, List, Optional, Tuple |
| 6 | +from osbot_utils.type_safe.Type_Safe import Type_Safe |
| 7 | +from osbot_utils.type_safe.primitives.safe_str.Safe_Str import Safe_Str |
| 8 | +from osbot_utils.type_safe.primitives.safe_int.Safe_Int import Safe_Int |
| 9 | +from osbot_utils.type_safe.primitives.safe_str.filesystem.Safe_Str__File__Path import Safe_Str__File__Path |
| 10 | +from osbot_utils.type_safe.primitives.safe_str.git.Safe_Str__Version import Safe_Str__Version |
| 11 | + |
| 12 | + |
| 13 | +class IR_Server(Type_Safe): |
| 14 | + url : Safe_Str |
| 15 | + variables : dict |
| 16 | + |
| 17 | +class IR_Parameter(Type_Safe): |
| 18 | + name : Safe_Str |
| 19 | + in_ : Safe_Str # path | query | header | cookie |
| 20 | + required : bool = False |
| 21 | + schema : dict = None # raw schema node |
| 22 | + |
| 23 | +class IR_RequestBody(Type_Safe): |
| 24 | + content_types : List[str] |
| 25 | + schema : dict = None |
| 26 | + |
| 27 | +class IR_Response(Type_Safe): |
| 28 | + status_code : Safe_Str |
| 29 | + content : dict = None |
| 30 | + |
| 31 | +class IR_Operation(Type_Safe): |
| 32 | + operation_id : Safe_Str |
| 33 | + method : Safe_Str |
| 34 | + path : Safe_Str__File__Path |
| 35 | + tag : Safe_Str = None |
| 36 | + summary : Safe_Str = None |
| 37 | + parameters : List[IR_Parameter] |
| 38 | + request_body : IR_RequestBody = None |
| 39 | + responses : List[IR_Response] |
| 40 | + deprecated : bool = False |
| 41 | + security : list = None |
| 42 | + |
| 43 | +class IR_Spec(Type_Safe): |
| 44 | + title : Safe_Str |
| 45 | + version : Safe_Str__Version = None |
| 46 | + servers : List[IR_Server] |
| 47 | + operations : List[IR_Operation] |
| 48 | + spec_hash : Safe_Str |
| 49 | + |
| 50 | +class OpenAPI__To__Python(Type_Safe): |
| 51 | + prefer_tag_sections : bool = True # group methods by tag as comments |
| 52 | + use_safe_primitives : bool = True # future hook for richer type mapping |
| 53 | + class_name_prefix : Safe_Str = 'Client__' # final class name is Client__{Service} |
| 54 | + default_timeout_sec : Safe_Int = 30 |
| 55 | + |
| 56 | + # ------------- public API ------------- |
| 57 | + |
| 58 | + def generate_from_json_str(self, json_str: str) -> str: |
| 59 | + spec_dict = json.loads(json_str) |
| 60 | + return self.generate_from_dict(spec_dict) |
| 61 | + |
| 62 | + def generate_from_app(self, app) -> str: |
| 63 | + """Convenience: accept a FastAPI app instance and use its in-memory OpenAPI.""" |
| 64 | + openapi_dict = app.openapi() |
| 65 | + return self.generate_from_dict(openapi_dict) |
| 66 | + |
| 67 | + def generate_from_dict(self, spec: dict) -> str: |
| 68 | + ir = self._build_ir(spec) |
| 69 | + content = self._render_client(ir, raw_spec=spec) |
| 70 | + return content |
| 71 | + |
| 72 | + # ------------- IR building ------------- |
| 73 | + |
| 74 | + def _build_ir(self, spec: dict) -> IR_Spec: |
| 75 | + title = Safe_Str (spec.get('info', {}).get('title', 'Service')) |
| 76 | + version = Safe_Str__Version(spec.get('info', {}).get('version', '0.0.0')) |
| 77 | + servers = [IR_Server(url=Safe_Str(s.get('url', '/')), variables=s.get('variables', {})) |
| 78 | + for s in spec.get('servers', [])] or [IR_Server(url=Safe_Str('/'))] |
| 79 | + |
| 80 | + # Compute stable hash of normalized spec |
| 81 | + spec_hash = hashlib.sha1(json.dumps(spec, sort_keys=True, separators=(',', ':')).encode()).hexdigest() |
| 82 | + |
| 83 | + ops: List[IR_Operation] = [] |
| 84 | + paths = spec.get('paths', {}) |
| 85 | + for path, path_item in paths.items(): |
| 86 | + for method in ('get','post','put','delete','patch','head','options','trace'): |
| 87 | + if method not in path_item: |
| 88 | + continue |
| 89 | + op_node = path_item[method] |
| 90 | + |
| 91 | + op_id = self._op_id(path, method, op_node.get('operationId')) |
| 92 | + tag = (op_node.get('tags') or [None])[0] |
| 93 | + params = self._collect_parameters(path_item, op_node) |
| 94 | + request_b = self._collect_request_body(op_node) |
| 95 | + responses = self._collect_responses(op_node.get('responses', {})) |
| 96 | + deprecated = bool(op_node.get('deprecated', False)) |
| 97 | + security = op_node.get('security') |
| 98 | + |
| 99 | + ops.append(IR_Operation( operation_id = Safe_Str(op_id) , |
| 100 | + method = Safe_Str(method.upper()) , |
| 101 | + path = Safe_Str__File__Path(path) , |
| 102 | + tag = Safe_Str(tag) if tag else None , |
| 103 | + summary = Safe_Str(op_node.get('summary')) if op_node.get('summary') else None, |
| 104 | + parameters = params , |
| 105 | + request_body = request_b , |
| 106 | + responses = responses , |
| 107 | + deprecated = deprecated , |
| 108 | + security = security )) |
| 109 | + |
| 110 | + return IR_Spec(title=title, version=version, servers=servers, operations=ops, spec_hash=Safe_Str(spec_hash)) |
| 111 | + |
| 112 | + def _op_id(self, path: str, method: str, explicit: Optional[str]) -> str: |
| 113 | + if explicit: |
| 114 | + return self._sanitize_method_name(explicit) |
| 115 | + # synthesize: get_/config/status -> get_config_status |
| 116 | + name = f"{method}_{re.sub(r'[^a-zA-Z0-9]+', '_', path).strip('_')}" |
| 117 | + return self._sanitize_method_name(name) |
| 118 | + |
| 119 | + def _sanitize_method_name(self, name: str) -> str: |
| 120 | + name = re.sub(r'[^a-zA-Z0-9_]', '_', name) |
| 121 | + if not re.match(r'[A-Za-z_]', name): |
| 122 | + name = f"op_{name}" |
| 123 | + return re.sub(r'__+', '_', name).lower() |
| 124 | + |
| 125 | + def _collect_parameters(self, path_item: dict, op_node: dict) -> List[IR_Parameter]: |
| 126 | + all_params = [] |
| 127 | + for source in (path_item.get('parameters', []), op_node.get('parameters', [])): |
| 128 | + for p in source: |
| 129 | + all_params.append(IR_Parameter(name=Safe_Str(p.get('name')), |
| 130 | + in_=Safe_Str(p.get('in')), |
| 131 | + required=bool(p.get('required', False)), |
| 132 | + schema=p.get('schema'))) |
| 133 | + # Deduplicate by (name, in) |
| 134 | + dedup: Dict[Tuple[str,str], IR_Parameter] = {} |
| 135 | + for p in all_params: |
| 136 | + key = (str(p.name), str(p.in_)) |
| 137 | + dedup[key] = p |
| 138 | + return list(dedup.values()) |
| 139 | + |
| 140 | + def _collect_request_body(self, op_node: dict) -> Optional[IR_RequestBody]: |
| 141 | + rb = op_node.get('requestBody') |
| 142 | + if not rb: |
| 143 | + return None |
| 144 | + content = rb.get('content') or {} |
| 145 | + content_types = list(content.keys()) |
| 146 | + schema = None |
| 147 | + # Prefer application/json |
| 148 | + if 'application/json' in content: |
| 149 | + schema = content['application/json'].get('schema') |
| 150 | + elif content_types: |
| 151 | + schema = (content[content_types[0]] or {}).get('schema') |
| 152 | + return IR_RequestBody(content_types=content_types, schema=schema) |
| 153 | + |
| 154 | + def _collect_responses(self, responses: dict) -> List[IR_Response]: |
| 155 | + out: List[IR_Response] = [] |
| 156 | + for status, node in responses.items(): |
| 157 | + out.append(IR_Response(status_code=Safe_Str(status), content=node.get('content'))) |
| 158 | + return out |
| 159 | + |
| 160 | + |
| 161 | + def _render_client(self, ir: IR_Spec, raw_spec: dict) -> str: |
| 162 | + class_name = self._client_class_name(ir.title) |
| 163 | + timestamp = datetime.datetime.utcnow().isoformat() + 'Z' |
| 164 | + sections = [] |
| 165 | + |
| 166 | + # file header |
| 167 | + sections.append(self._render_header(ir, timestamp)) |
| 168 | + sections.append(self._render_imports()) |
| 169 | + sections.append(self._render_config_and_http_classes()) |
| 170 | + sections.append(self._render_client_class(class_name, ir)) |
| 171 | + |
| 172 | + return "\n".join(sections) |
| 173 | + |
| 174 | + def _client_class_name(self, title: str) -> str: |
| 175 | + name = re.sub(r'[^a-zA-Z0-9]+', '_', title).strip('_') |
| 176 | + if not name: |
| 177 | + name = 'Service' |
| 178 | + return f"{self.class_name_prefix}{name}" |
| 179 | + |
| 180 | + # ---- code snippets ---- |
| 181 | + |
| 182 | + def _render_header(self, ir: IR_Spec, timestamp: str) -> str: |
| 183 | + return f"""# Auto-generated client for {ir.title} |
| 184 | +# spec_version : {ir.version} |
| 185 | +# spec_hash : {ir.spec_hash} |
| 186 | +# generated_at : {timestamp} |
| 187 | +# generator : OpenAPI__To__Python""" |
| 188 | + |
| 189 | + def _render_imports(self) -> str: |
| 190 | + return """ |
| 191 | +import requests |
| 192 | +import urllib.parse |
| 193 | +from typing import Any, Dict, Optional |
| 194 | +from osbot_utils.type_safe.Type_Safe import Type_Safe |
| 195 | +from osbot_utils.type_safe.primitives.safe_int import Safe_Int |
| 196 | +from osbot_utils.type_safe.primitives.safe_str.web.Safe_Str__Url import Safe_Str__Url |
| 197 | +
|
| 198 | +""" |
| 199 | + |
| 200 | + def _render_config_and_http_classes(self) -> str: |
| 201 | + return """ |
| 202 | +class Fast_API__Client__Config(Type_Safe): |
| 203 | + base_url : Safe_Str__Url = None |
| 204 | + timeout_sec : Safe_Int = 30 |
| 205 | + bearer_token : str = None |
| 206 | + api_key_name : str = None |
| 207 | + api_key_value : str = None |
| 208 | + headers : Dict[str, str] |
| 209 | +
|
| 210 | +class Fast_API__Client__Http(Type_Safe): |
| 211 | + config: Fast_API__Client__Config |
| 212 | +
|
| 213 | + def headers(self) -> Dict[str, str]: |
| 214 | + headers = self.config.headers.copy() |
| 215 | + if self.config.bearer_token: |
| 216 | + headers['Authorization'] = f'Bearer {self.config.bearer_token}' |
| 217 | + if self.config.api_key_name and self.config.api_key_value: |
| 218 | + headers[str(self.config.api_key_name)] = str(self.config.api_key_value) |
| 219 | + return headers |
| 220 | +
|
| 221 | + def build_url(self, path: str, query: Dict[str, Any] = None) -> str: |
| 222 | + base = str(self.config.base_url).rstrip('/') |
| 223 | + path = '/' + path.lstrip('/') |
| 224 | + if query: |
| 225 | + q = {k: v for k, v in query.items() if v is not None} # Drop None values |
| 226 | + if q: |
| 227 | + return base + path + '?' + urllib.parse.urlencode(q, doseq=True) |
| 228 | + return base + path |
| 229 | +
|
| 230 | + def request_json(self, method: str , # HTTP method (GET, POST, etc.) |
| 231 | + path : str , # URL path |
| 232 | + query : Dict[str, Any] = None , # Query parameters |
| 233 | + body : Any = None # Request body |
| 234 | + ) -> Any : # JSON object or string response |
| 235 | + url = self.build_url(path, query) |
| 236 | + headers = self.headers() |
| 237 | +
|
| 238 | + response = requests.request(method = method , |
| 239 | + url = url , |
| 240 | + headers = headers, |
| 241 | + json = body if isinstance(body, (dict, list)) else None, |
| 242 | + data = body if not isinstance(body, (dict, list)) else None, |
| 243 | + timeout=int(self.config.timeout_sec)) |
| 244 | + response.raise_for_status() |
| 245 | +
|
| 246 | + if response.headers.get('Content-Type', '').startswith('application/json'): |
| 247 | + return response.json() |
| 248 | + return response.text |
| 249 | +""" |
| 250 | + |
| 251 | + def _render_client_class(self, class_name: str, ir: IR_Spec) -> str: |
| 252 | + methods_section = self._render_methods(ir) |
| 253 | + |
| 254 | + return f""" |
| 255 | +class {class_name}(Type_Safe): |
| 256 | + config : Fast_API__Client__Config |
| 257 | + http : Fast_API__Client__Http |
| 258 | +
|
| 259 | + def __init__(self, url:Safe_Str__Url=None, **kwargs): |
| 260 | + super().__init__(**kwargs) |
| 261 | + if url: |
| 262 | + self.config.url = url |
| 263 | + self.http = Fast_API__Client__Http(config=self.config) |
| 264 | +
|
| 265 | +{methods_section}""" |
| 266 | + |
| 267 | + def _render_methods(self, ir: IR_Spec) -> str: |
| 268 | + # Group by tag |
| 269 | + groups: Dict[str, List[IR_Operation]] = {} |
| 270 | + for op in ir.operations: |
| 271 | + tag = str(op.tag) if op.tag else 'default' |
| 272 | + groups.setdefault(tag, []).append(op) |
| 273 | + |
| 274 | + lines: List[str] = [] |
| 275 | + for tag in sorted(groups.keys()): |
| 276 | + if self.prefer_tag_sections: |
| 277 | + lines.append(f"\n # -------------------- tag: {tag} --------------------\n") |
| 278 | + for op in groups[tag]: |
| 279 | + lines.append(self._render_method(op)) |
| 280 | + return "".join(lines) |
| 281 | + |
| 282 | + def _render_method(self, op: IR_Operation) -> str: |
| 283 | + # Build signature - keeping it simple for the ping example |
| 284 | + sig_parts = ["self"] |
| 285 | + sig_parts.append("timeout_sec: Optional[int] = None") |
| 286 | + sig = ", ".join(sig_parts) |
| 287 | + |
| 288 | + # Build path (for simple case without parameters) |
| 289 | + path_expr = f"f'{op.path}'" |
| 290 | + |
| 291 | + # Method docstring |
| 292 | + summary = str(op.summary) if op.summary else f"{op.method} {op.path}" |
| 293 | + doc = f'""" {summary} """' |
| 294 | + |
| 295 | + return f""" |
| 296 | + def {op.operation_id}({sig}): |
| 297 | + {doc} |
| 298 | + if timeout_sec is not None: |
| 299 | + self.config.timeout_sec = timeout_sec |
| 300 | +
|
| 301 | + path = {path_expr} |
| 302 | + query = None |
| 303 | + return self.http.request_json(method='{op.method}', path=path, query=query) |
| 304 | +""" |
0 commit comments