import inspect import logging import re from typing import Any, Callable, Dict, List, Optional, Tuple, Union from flask import Flask, Blueprint from pydantic import BaseModel, create_model # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class EndpointData(BaseModel): """Data class to store endpoint information.""" route: str method: str parameters: Dict[str, str] request_body: Optional[Dict[str, Any]] = None response_type: str description: Optional[str] = None class APICataloger: """ A class to catalog API endpoints defined in a Flask application. """ def __init__(self, app: Flask): """ Initializes the APICataloger with a Flask application. Args: app: The Flask application to catalog. """ self.app = app self.endpoints: List[EndpointData] = [] def _extract_parameters(self, route: str) -> Dict[str, str]: """ Extracts parameters from a route string. Args: route: The route string (e.g., '/users/'). Returns: A dictionary of parameter names and types. """ parameters: Dict[str, str] = {} matches = re.findall(r'<(?:[^:]+:)?([^>]+)>', route) # Matches or for match in matches: param_name = match param_type = "string" # Default to string if no type is specified in the route if ":" in match: param_type = match.split(":")[0] # Extract the type if provided parameters[param_name] = param_type return parameters def _extract_request_body(self, func: Callable) -> Optional[Dict[str, Any]]: """ Extracts request body schema from the function's signature. Assumes the request body is a Pydantic BaseModel. Args: func: The view function. Returns: A dictionary representing the request body schema, or None if not found. """ signature = inspect.signature(func) for param in signature.parameters.values(): if isinstance(param.annotation, type) and issubclass(param.annotation, BaseModel): try: # Convert Pydantic model to a JSON schema dictionary return param.annotation.model_json_schema() except Exception as e: logger.warning(f"Failed to extract request body schema: {e}") return None return None def catalog_endpoints(self) -> None: """ Catalogs all API endpoints in the Flask application. """ for rule in self.app.url_map.iter_rules(): # Exclude static routes if rule.endpoint == 'static': continue methods = rule.methods if methods is None: continue # Should not happen, but for safety view_func = self.app.view_functions[rule.endpoint] for method in methods: if method in ['HEAD', 'OPTIONS']: continue parameters = self._extract_parameters(str(rule)) request_body = self._extract_request_body(view_func) # Attempt to extract response type from docstring docstring = inspect.getdoc(view_func) response_type = "application/json" # Default response type description = None if docstring: lines = docstring.splitlines() for line in lines: if line.strip().startswith("Returns:"): try: response_type_match = re.search(r"Returns:\s*(\w+)", line) if response_type_match: response_type = response_type_match.group(1) else: response_type = "application/json" # Default to json if can't parse except Exception as e: logger.warning(f"Error parsing response type from docstring: {e}") response_type = "application/json" elif not description: description = docstring.strip() endpoint_data = EndpointData( route=str(rule), method=method, parameters=parameters, request_body=request_body, response_type=response_type, description=description ) self.endpoints.append(endpoint_data) logger.info(f"Cataloged endpoint: {method} {rule}") def generate_openapi_fragment(self) -> Dict[str, Any]: """ Generates an OpenAPI spec fragment from the cataloged endpoints. Returns: A dictionary representing the OpenAPI spec fragment. """ openapi_fragment: Dict[str, Any] = {"paths": {}} for endpoint in self.endpoints: path = endpoint.route method = endpoint.method.lower() if path not in openapi_fragment["paths"]: openapi_fragment["paths"][path] = {} openapi_fragment["paths"][path][method] = { "summary": endpoint.description or f"{method.upper()} {path}", # Use description if available "parameters": [ { "name": param_name, "in": "path", "required": True, "schema": {"type": param_type} } for param_name, param_type in endpoint.parameters.items() ], "responses": { "200": { # Assuming 200 OK for simplicity "description": "Successful response", "content": { "application/json": { "schema": {"type": "object"} # Placeholder, refine as needed } } } } } if endpoint.request_body: openapi_fragment["paths"][path][method]["requestBody"] = { "content": { "application/json": { "schema": endpoint.request_body } } } return openapi_fragment def generate_markdown(self) -> str: """ Generates a markdown representation of the API catalog. Returns: A string containing the markdown representation. """ markdown = "# API Catalog\n\n" for endpoint in self.endpoints: markdown += f"## {endpoint.method} {endpoint.route}\n\n" if endpoint.description: markdown += f"{endpoint.description}\n\n" markdown += "**Parameters:**\n\n" if endpoint.parameters: for name, type in endpoint.parameters.items(): markdown += f"* `{name}` ({type})\n" else: markdown += "None\n" if endpoint.request_body: markdown += "\n**Request Body:**\n\n" markdown += "json\n" markdown += str(endpoint.request_body) markdown += "\n\n" markdown += f"\n**Response Type:** {endpoint.response_type}\n\n" return markdown if __name__ == '__main__': # Example usage: app = Flask(__name__) # Define a Pydantic model for request body example class UserCreate(BaseModel): username: str email: str age: int @app.route('/users/', methods=['GET']) def get_user(user_id: int): """ Retrieves a user by ID. Args: user_id: The ID of the user to retrieve. Returns: application/json """ return {'user_id': user_id} @app.route('/users', methods=['POST']) def create_user(user: UserCreate): """ Creates a new user. Args: user: The user data. Returns: application/json """ return {'message': 'User created successfully'} @app.route('/items/', methods=['GET']) def get_item(item_name: str): """ Retrieves an item by name. Args: item_name: The name of the item to retrieve. Returns: application/json """ return {'item_name': item_name} cataloger = APICataloger(app) cataloger.catalog_endpoints() openapi_fragment = cataloger.generate_openapi_fragment() print("OpenAPI Fragment:\n", openapi_fragment) markdown_output = cataloger.generate_markdown() print("\nMarkdown Output:\n", markdown_output)