Initial upload: API Mock CLI v0.1.0

src/core/server.py

@@ -0,0 +1,130 @@
+import asyncio
+import signal
+import sys
+from typing import Optional, Dict, Any
+from contextlib import asynccontextmanager
+import uvicorn
+from fastapi import FastAPI, Request, Response
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+from src.core.matcher import Matcher
+from src.core.templating import TemplatingEngine
+from src.core.validator import Validator
+from src.core.tunnel import Tunnel, TunnelError
+from src.models.endpoint import Endpoint
+from src.models.config import ServerConfig
+from src.models.request import RequestValidation
+
+
+class MockServer:
+    def __init__(self, config: ServerConfig):
+        self.config = config
+        self.matcher = Matcher()
+        self.templating = TemplatingEngine()
+        self.tunnel: Optional[Tunnel] = None
+        self._server_task: Optional[asyncio.Task] = None
+        self._shutdown_event = asyncio.Event()
+
+    def register_endpoint(self, endpoint: Endpoint) -> None:
+        if endpoint.enabled:
+            self.matcher.add_endpoint(endpoint)
+
+    def register_endpoints(self, endpoints: list[Endpoint]) -> None:
+        for endpoint in endpoints:
+            self.register_endpoint(endpoint)
+
+    def _build_response(self, endpoint: Endpoint, context: Dict[str, Any]) -> Dict[str, Any]:
+        response_config = endpoint.response
+        status_code = response_config.get("status_code", 200)
+        headers = response_config.get("headers", {})
+        body = response_config.get("body")
+        if body:
+            body = self.templating.render_dict(body, context)
+        return {"status_code": status_code, "headers": headers, "body": body}
+
+    async def _handle_request(self, request: Request, endpoint: Endpoint, path_params: Dict[str, str]) -> Response:
+        try:
+            body = None
+            if request.method in ["POST", "PUT", "PATCH"]:
+                body = await request.json()
+            query_params = dict(request.query_params)
+            headers = dict(request.headers)
+            headers.pop("host", None)
+            context = self.templating.build_context(
+                path_params=path_params,
+                query_params=query_params,
+                headers=headers,
+                body=body
+            )
+            if endpoint.validators:
+                validation_rules = RequestValidation(**endpoint.validators)
+                validator = Validator(validation_rules)
+                valid, errors = validator.validate(body=body, query=query_params, headers=headers, path=path_params)
+                if not valid:
+                    return JSONResponse(
+                        status_code=400,
+                        content={"validation_error": validator.format_errors(errors)}
+                    )
+            response_data = self._build_response(endpoint, context)
+            return JSONResponse(
+                status_code=response_data["status_code"],
+                content=response_data["body"],
+                headers=response_data["headers"]
+            )
+        except Exception as e:
+            return JSONResponse(
+                status_code=500,
+                content={"error": str(e)}
+            )
+
+    def _create_app(self) -> FastAPI:
+        @asynccontextmanager
+        async def lifespan(app: FastAPI):
+            if not self.config.offline and self.config.tunnel:
+                try:
+                    tunnel = Tunnel(self.config.ngrok_authtoken)
+                    public_url = tunnel.start(self.config.port)
+                    print(f"Public URL: {public_url}")
+                    self.tunnel = tunnel
+                except TunnelError as e:
+                    print(f"Tunnel creation failed: {e}. Running in offline mode.")
+            yield
+            if self.tunnel:
+                self.tunnel.stop()
+        app = FastAPI(lifespan=lifespan)
+        app.add_middleware(
+            CORSMiddleware,
+            allow_origins=self.config.cors_origins or ["*"],
+            allow_credentials=True,
+            allow_methods=["*"],
+            allow_headers=["*"],
+        )
+        async def dynamic_handler(request: Request):
+            path = request.url.path
+            method = request.method
+            match = self.matcher.match(path, method)
+            if match:
+                endpoint, path_params = match
+                return await self._handle_request(request, endpoint, path_params)
+            return JSONResponse(status_code=404, content={"error": "Not Found"})
+        for method in ["GET", "POST", "PUT", "DELETE", "PATCH"]:
+            app.add_api_route("/{path:path}", dynamic_handler, methods=[method])
+        return app
+
+    async def start(self) -> None:
+        app = self._create_app()
+        config = uvicorn.Config(
+            app,
+            host=self.config.host,
+            port=self.config.port,
+            reload=self.config.reload,
+            log_level=self.config.log_level
+        )
+        server = uvicorn.Server(config)
+        await server.serve()
+
+    def run(self) -> None:
+        try:
+            asyncio.run(self.start())
+        except KeyboardInterrupt:
+            print("Shutting down...")