api tested and completed

This commit is contained in:
2025-04-02 14:34:14 +03:00
parent 9e27893e8f
commit 27c48bb86a
30 changed files with 2695 additions and 24 deletions

19
ApiServices/README.md Normal file
View File

@@ -0,0 +1,19 @@
Account Create
Endpoint('/account/create') -> ClassVar[AccountCreate]
AccountCreate.related_callable -> Serves Result
* From Endpoint route to related_callable
AccountCreate:
EndpointURL('http://accountservice/account/create')
def related_callable():
callable1 -> UUID[] -> Result
callable2 -> UUID[]
callable3 -> UUID[]
callable4 -> UUID[]
callable5 -> UUID[]
ApplicationService -> Serves only app pages that are reachable for Client Side

View File

@@ -15,11 +15,12 @@ RUN poetry config virtualenvs.create false \
&& pip cache purge && rm -rf ~/.cache/pypoetry
# Copy application code
COPY /ApiServices .
COPY /ApiServices/TemplateService /ApiServices/TemplateService
COPY /Controllers /Controllers
COPY /Schemas /Schemas
# Set Python path to include app directory
ENV PYTHONPATH=/ PYTHONUNBUFFERED=1 PYTHONDONTWRITEBYTECODE=1
# Run the application using the configured uvicorn server
CMD ["poetry", "run", "python", "app.py"]
CMD ["poetry", "run", "python", "ApiServices/TemplateService/app.py"]

View File

View File

@@ -0,0 +1,15 @@
import uvicorn
from ApiServices.TemplateService.config import template_api_config
from ApiServices.TemplateService.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(**template_api_config.app_as_dict)
uvicorn.Server(uvicorn_config).run()

View File

@@ -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_")
template_api_config = Configs()

View File

@@ -0,0 +1,41 @@
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import RedirectResponse
from fastapi.staticfiles import StaticFiles
from ApiServices.TemplateService.create_route import RouteRegisterController
from ApiServices.TemplateService.endpoints.routes import get_routes
from ApiServices.TemplateService.open_api_creator import create_openapi_schema
from ApiServices.TemplateService.middlewares.token_middleware import token_middleware
from ApiServices.TemplateService.config import template_api_config
def create_app():
application = FastAPI(**template_api_config.api_info)
# application.mount(
# "/application/static",
# StaticFiles(directory="application/static"),
# name="static",
# )
application.add_middleware(
CORSMiddleware,
allow_origins=template_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()
application.openapi = lambda _=application: create_openapi_schema(_)
return application

View File

@@ -0,0 +1,14 @@
from typing import List
from fastapi import APIRouter, FastAPI
class RouteRegisterController:
def __init__(self, app: FastAPI, router_list: List[APIRouter]):
self.router_list = router_list
self.app = app
def register_routes(self):
for router in self.router_list:
self.app.include_router(router)
return self.app

View File

@@ -0,0 +1,17 @@
from fastapi import APIRouter
def get_routes() -> list[APIRouter]:
return []
def get_safe_endpoint_urls() -> list[tuple[str, str]]:
return [
("/", "GET"),
("/docs", "GET"),
("/redoc", "GET"),
("/openapi.json", "GET"),
("/auth/register", "POST"),
("/auth/login", "POST"),
("/metrics", "GET"),
]

View File

@@ -0,0 +1,21 @@
from fastapi import Request, Response
def get_safe_endpoint_urls() -> list:
return []
async def token_middleware(request: Request, call_next):
# from application.routes.routes import get_safe_endpoint_urls
base_url = "/".join(request.url.path.split("/")[:3])
safe_endpoints = [_[0] for _ in get_safe_endpoint_urls()]
if base_url in safe_endpoints:
return await call_next(request)
token = request.headers.get("Authorization")
if not token:
return Response(content="Missing token", status_code=400)
response = await call_next(request)
return response

View File

@@ -0,0 +1,89 @@
from typing import Any, Dict
from fastapi import FastAPI
from fastapi.routing import APIRoute
from fastapi.openapi.utils import get_openapi
from ApiServices.TemplateService.config import template_api_config
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
def create_security_schemes(self) -> 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 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,
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()

View File

@@ -1,9 +0,0 @@
from Controllers.Postgres.config import postgres_configs
from Controllers.Mongo.config import mongo_configs
from Controllers.Mongo.implementations import run_all_tests
if __name__ == "__main__":
print(f"Hello from the Test Service {mongo_configs.url}")
print(f"Hello from the Test Service {postgres_configs.url}")
run_all_tests()