updated event and router stacks

ApiDefaults/app.py

@@ -0,0 +1,16 @@
+import uvicorn
+
+from ApiDefaults.config import api_config
+from ApiDefaults.create_app import create_app
+
+# from prometheus_fastapi_instrumentator import Instrumentator
+
+
+app = create_app()  # Create FastAPI application
+# Instrumentator().instrument(app=app).expose(app=app)          # Setup Prometheus metrics
+
+
+if __name__ == "__main__":
+    # Run the application with Uvicorn Server
+    uvicorn_config = uvicorn.Config(**api_config.app_as_dict)
+    uvicorn.Server(uvicorn_config).run()

ApiDefaults/config.py

@@ -0,0 +1,64 @@
+from pydantic_settings import BaseSettings, SettingsConfigDict
+from fastapi.responses import JSONResponse
+
+
+class Configs(BaseSettings):
+    """
+    ApiTemplate configuration settings.
+    """
+
+    PATH: str = ""
+    HOST: str = ""
+    PORT: int = 0
+    LOG_LEVEL: str = "info"
+    RELOAD: int = 0
+    ACCESS_TOKEN_TAG: str = ""
+
+    ACCESS_EMAIL_EXT: str = ""
+    TITLE: str = ""
+    ALGORITHM: str = ""
+    ACCESS_TOKEN_LENGTH: int = 90
+    REFRESHER_TOKEN_LENGTH: int = 144
+    EMAIL_HOST: str = ""
+    DATETIME_FORMAT: str = ""
+    FORGOT_LINK: str = ""
+    ALLOW_ORIGINS: list = ["http://localhost:3000"]
+    VERSION: str = "0.1.001"
+    DESCRIPTION: str = ""
+
+    @property
+    def app_as_dict(self) -> dict:
+        """
+        Convert the settings to a dictionary.
+        """
+        return {
+            "app": self.PATH,
+            "host": self.HOST,
+            "port": int(self.PORT),
+            "log_level": self.LOG_LEVEL,
+            "reload": bool(self.RELOAD),
+        }
+
+    @property
+    def api_info(self):
+        """
+        Returns a dictionary with application information.
+        """
+        return {
+            "title": self.TITLE,
+            "description": self.DESCRIPTION,
+            "default_response_class": JSONResponse,
+            "version": self.VERSION,
+        }
+
+    @classmethod
+    def forgot_link(cls, forgot_key):
+        """
+        Generate a forgot password link.
+        """
+        return cls.FORGOT_LINK + forgot_key
+
+    model_config = SettingsConfigDict(env_prefix="API_")
+
+
+api_config = Configs()

ApiDefaults/create_app.py

@@ -0,0 +1,58 @@
+from ApiControllers.abstracts.event_clusters import RouterCluster, EventCluster
+from fastapi import FastAPI, Request
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import RedirectResponse
+
+
+def create_events_if_any_cluster_set():
+    import Events 
+    if not Events.__all__:
+        return
+    
+    router_cluster_stack: list[RouterCluster] = [getattr(Events, e, None) for e in Events.__all__]
+    for router_cluster in router_cluster_stack:
+        event_cluster_stack: list[EventCluster] = list(router_cluster.event_clusters.values())
+        for event_cluster in event_cluster_stack:
+            print(f"Creating event:", event_cluster.name)
+            try:
+                event_cluster.set_events_to_database()
+            except Exception as e:
+                print(f"Error creating event cluster: {e}")
+
+
+def create_app():
+    from ApiDefaults.open_api_creator import create_openapi_schema
+    from ApiDefaults.config import api_config
+
+    from ApiControllers.middlewares.token_middleware import token_middleware
+    from ApiControllers.initializer.create_route import RouteRegisterController
+
+    from Endpoints.routes import get_routes
+
+    application = FastAPI(**api_config.api_info)
+    # application.mount(
+    #     "/application/static",
+    #     StaticFiles(directory="application/static"),
+    #     name="static",
+    # )
+    application.add_middleware(
+        CORSMiddleware,
+        allow_origins=api_config.ALLOW_ORIGINS,
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+    @application.middleware("http")
+    async def add_token_middleware(request: Request, call_next):
+        return await token_middleware(request, call_next)
+
+    @application.get("/", description="Redirect Route", include_in_schema=False)
+    async def redirect_to_docs():
+        return RedirectResponse(url="/docs")
+
+    route_register = RouteRegisterController(app=application, router_list=get_routes())
+    application = route_register.register_routes()
+    create_events_if_any_cluster_set()
+    application.openapi = lambda _=application: create_openapi_schema(_)
+    return application

ApiDefaults/open_api_creator.py

@@ -0,0 +1,116 @@
+from typing import Any, Dict
+from fastapi import FastAPI
+from fastapi.routing import APIRoute
+from fastapi.openapi.utils import get_openapi
+
+from ApiDefaults.config import api_config as template_api_config
+from Endpoints.routes import get_safe_endpoint_urls
+
+
+class OpenAPISchemaCreator:
+    """
+    OpenAPI schema creator and customizer for FastAPI applications.
+    """
+
+    def __init__(self, app: FastAPI):
+        """
+        Initialize the OpenAPI schema creator.
+
+        Args:
+            app: FastAPI application instance
+        """
+        self.app = app
+        self.safe_endpoint_list: list[tuple[str, str]] = get_safe_endpoint_urls()
+        self.routers_list = self.app.routes
+
+    @staticmethod
+    def create_security_schemes() -> Dict[str, Any]:
+        """
+        Create security scheme definitions.
+
+        Returns:
+            Dict[str, Any]: Security scheme configurations
+        """
+
+        return {
+            "BearerAuth": {
+                "type": "apiKey",
+                "in": "header",
+                "name": template_api_config.ACCESS_TOKEN_TAG,
+                "description": "Enter: **'Bearer <JWT>'**, where JWT is the access token",
+            }
+        }
+
+    def configure_route_security(
+        self, path: str, method: str, schema: Dict[str, Any]
+    ) -> None:
+        """
+        Configure security requirements for a specific route.
+
+        Args:
+            path: Route path
+            method: HTTP method
+            schema: OpenAPI schema to modify
+        """
+        if not schema.get("paths", {}).get(path, {}).get(method):
+            return
+
+        # Check if endpoint is in safe list
+        endpoint_path = f"{path}:{method}"
+        list_of_safe_endpoints = [
+            f"{e[0]}:{str(e[1]).lower()}" for e in self.safe_endpoint_list
+        ]
+        if endpoint_path not in list_of_safe_endpoints:
+            if "security" not in schema["paths"][path][method]:
+                schema["paths"][path][method]["security"] = []
+            schema["paths"][path][method]["security"].append({"BearerAuth": []})
+
+    def create_schema(self) -> Dict[str, Any]:
+        """
+        Create the complete OpenAPI schema.
+
+        Returns:
+            Dict[str, Any]: Complete OpenAPI schema
+        """
+        openapi_schema = get_openapi(
+            title=template_api_config.TITLE,
+            description=template_api_config.DESCRIPTION,
+            version=template_api_config.VERSION,
+            routes=self.app.routes,
+        )
+
+        # Add security schemes
+        if "components" not in openapi_schema:
+            openapi_schema["components"] = {}
+
+        openapi_schema["components"]["securitySchemes"] = self.create_security_schemes()
+
+        # Configure route security and responses
+        for route in self.app.routes:
+            if isinstance(route, APIRoute) and route.include_in_schema:
+                path = str(route.path)
+                methods = [method.lower() for method in route.methods]
+                for method in methods:
+                    self.configure_route_security(path, method, openapi_schema)
+
+        # Add custom documentation extensions
+        openapi_schema["x-documentation"] = {
+            "postman_collection": "/docs/postman",
+            "swagger_ui": "/docs",
+            "redoc": "/redoc",
+        }
+        return openapi_schema
+
+
+def create_openapi_schema(app: FastAPI) -> Dict[str, Any]:
+    """
+    Create OpenAPI schema for a FastAPI application.
+
+    Args:
+        app: FastAPI application instance
+
+    Returns:
+        Dict[str, Any]: Complete OpenAPI schema
+    """
+    creator = OpenAPISchemaCreator(app)
+    return creator.create_schema()