import subprocess import time from collections.abc import Callable from contextlib import asynccontextmanager from fastapi import FastAPI, HTTPException, Request, Response from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from fastapi.staticfiles import StaticFiles from sqlalchemy.exc import ProgrammingError from fastapi import Depends from fastapi.security import OAuth2PasswordBearer #from app.api.auth.router import auth_router # Import the authentication router from app.api.user.router import user_router from app.api.brand.router import brand_router from app.api.segment.router import segment_router from app.api.store_formate.router import store_formate_router from app.api.student.router import student_router from app.api.simulation.router import simulation_router from app.api.group.router import group_router from app.api.customer_location.router import customer_location_router from app.api.catchment_potential.router import catchment_potential_router from app.api.network.router import network_planning_router from app.api.payment.router import payment_router from app.api.excel_download.router import excel_router from app.api.civil.router import civil_router from app.api.display_racking_create.router import display_racking_unit_router from app.api.carpentry.router import carpentry_furniture_router from app.api.commercial_equipment.router import commercial_equipment_router from app.api.plumbing.router import plumbing_router from app.api.trading_month.router import trading_month_router from app.api.pre_operating_expenses.router import pre_operating_router from app.api.competitor_analysis.router import competitor_analysis_router from app.config import settings # Import configuration settings from app.database.main.mongo import ( MongoDBSingleton, # Import MongoDB singleton instance ) from sqlalchemy.exc import OperationalError from app.exceptions import ( handle_generic_exception_handler, handle_programming_error, http_exception_error_handler, handel_operational_error_handler, handel_attribute_error_handler ) from app.utils.schemas_utils import CustomResponse mongodb_instance = MongoDBSingleton() # Initialize MongoDB singleton instance @asynccontextmanager async def lifespan(_app: FastAPI): """Initialize the application.""" # Load the MongoDB instance _ = mongodb_instance yield # Yield control back to the application app = FastAPI( title=settings.PROJECT_NAME, version=settings.API_VERSION, description=settings.DESCRIPTION, # docs_url=None, # Optionally disable the interactive API docs # redoc_url=None, # Optionally disable the ReDoc docs lifespan=lifespan, # Set the lifespan context manager debug=True, # Enable debug mode ) @app.middleware("http") async def add_process_time_header(request: Request, call_next: Callable) -> Response: """ Add process time header to the response. Args: request (Request): The incoming request. call_next (Callable): The next callable in the middleware chain. Returns: Response: The response with added process time header. """ start_time = time.time() response = await call_next(request) # Process the request process_time = time.time() - start_time response.headers["X-Process-Time"] = str(process_time) # Add process time header to the response return response # Add CORS middleware to the application app.add_middleware( CORSMiddleware, allow_origins=["*"], # Allow all origins (set to a list of allowed origins for production) allow_credentials=True, allow_methods=["*"], # Allow all HTTP methods allow_headers=["*"], # Allow all HTTP headers expose_headers=["*"], # Expose all headers ) # Include the authentication router with prefix and tags #app.include_router(auth_router, prefix="/FASTAPI", tags=["AUTH"]) app.include_router(user_router, prefix="/FASTAPI", tags=["User"]) app.include_router(brand_router, prefix="/FASTAPI", tags=["Brand"]) app.include_router(segment_router, prefix="/FASTAPI", tags=["Segment"]) app.include_router(store_formate_router, prefix="/FASTAPI", tags=["Store Formate"]) app.include_router(student_router, prefix="/FASTAPI", tags=["Student"]) app.include_router(simulation_router,prefix="/FASTAPI", tags=["Simulation"]) app.include_router(group_router,prefix="/FASTAPI", tags=["Group"]) app.include_router(customer_location_router,prefix="/FASTAPI", tags=["Customer Location"]) app.include_router(catchment_potential_router,prefix="/FASTAPI", tags=["Catchment Potential"]) app.include_router(network_planning_router,prefix="/FASTAPI", tags=["Network Planning"]) app.include_router(payment_router,prefix="/FASTAPI", tags=["Payment"]) app.include_router(excel_router,prefix="/FASTAPI", tags=["Download Excel"]) app.include_router(civil_router,prefix="/FASTAPI", tags=["Civil and Backroom Work"]) app.include_router(display_racking_unit_router,prefix="/FASTAPI", tags=["Display and Racking Units"]) app.include_router(carpentry_furniture_router,prefix="/FASTAPI", tags=["Carpentry and Furniture"]) app.include_router(commercial_equipment_router,prefix="/FASTAPI", tags=["Commercial Equipment"]) app.include_router(plumbing_router,prefix="/FASTAPI", tags=["Plumbing"]) app.include_router(trading_month_router,prefix="/FASTAPI", tags=["Trading Months"]) app.include_router(pre_operating_router,prefix="/FASTAPI", tags=["Pre-Operating Expenses"]) app.include_router(competitor_analysis_router,prefix="/FASTAPI", tags=["Competitor Analysis"]) # HTTP Exception Handler @app.exception_handler(HTTPException) async def exception_handler(request: Request, exc: HTTPException) -> JSONResponse: """Handle HTTP exceptions.""" return http_exception_error_handler(request, exc) # Use custom HTTP exception handler # MySQL Exception Handler @app.exception_handler(ProgrammingError) async def programming_error_handler(request: Request, exc: ProgrammingError) -> JSONResponse: """Handle MySQL ProgrammingError exceptions.""" return await handle_programming_error(request, exc) # Use custom programming error handler # General Exceptions @app.exception_handler(Exception) async def generic_exception_handler(request: Request, exc: Exception) -> JSONResponse: """Handle generic exceptions.""" return await handle_generic_exception_handler(request, exc) # Use custom generic exception handler # Operational Errors @app.exception_handler(OperationalError) async def operational_error_handler(request: Request, exc: OperationalError) -> HTTPException: raise await handel_operational_error_handler(request, exc) # Attribute Errors @app.exception_handler(AttributeError) async def attribute_error_handler(request: Request, exc: AttributeError)-> HTTPException: return await handel_attribute_error_handler(request, exc) # Run Migrations @app.get("/migrate",response_model_exclude_none=True) async def run_migrations(message: str)-> CustomResponse: try: # subprocess.run(['alembic', 'downgrade', ''], shell=True) # Run Alembic revision command subprocess.run(['alembic', 'revision', '--autogenerate', '-m', '"{}"'.format(message)]) # Run Alembic upgrade command subprocess.run( ["alembic", "upgrade", "head"] ) return CustomResponse(status="1", message="Migration successful") except subprocess.CalledProcessError as e: raise HTTPException(status_code=500, detail=str(e)) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) app.mount("/uploads",StaticFiles(directory="uploads"),name="uploads") # Create OAuth2PasswordBearer instances for both user and student logins #user_oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/FASTAPI/user/login") student_oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/FASTAPI/student/login") # Manually define the OpenAPI schema to include both OAuth2 flows (User and Student) @app.on_event("startup") def on_startup(): app.openapi_schema = { "openapi": "3.0.0", "info": { "title": "My API", "version": "1.0.0", "description": "API Documentation with User and Student Login", }, "paths": { "/some-protected-route": { "get": { "security": [ {"user_oauth2": []}, {"student_oauth2": []}, ], # Both flows (User and Student) are shown here "operationId": "getSomeProtectedRoute", "summary": "A protected route for both user and student", "responses": { "200": {"description": "Successful response"}, }, } }, }, "components": { "securitySchemes": { # "user_oauth2": { # "type": "oauth2", # "flow": { # "password": { # "tokenUrl": "/FASTAPI/user/login", # User login token URL # } # } # }, "student_oauth2": { "type": "oauth2", "flow": { "password": { "tokenUrl": "/FASTAPI/student/login", # Student login token URL } } }, }, } } #Define a protected route that can be accessed with either user or student login # @app.get("/some-protected-route") # def get_some_protected_route( # user: str = Depends(user_oauth2_scheme), # Use user OAuth2 token # ): # return {"message": "This is a protected route", "user": user} # Define a student-specific route @app.get("/student-protected-route") def get_student_protected_route( student: str = Depends(student_oauth2_scheme), # Use student OAuth2 token ): return {"message": "This is a protected route for students", "student": student}