106 lines
No EOL
3.9 KiB
Python
106 lines
No EOL
3.9 KiB
Python
from fastapi import FastAPI, Request
|
|
from fastapi.responses import JSONResponse, FileResponse
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
import os
|
|
import csv
|
|
import json
|
|
from pathlib import Path
|
|
|
|
app = FastAPI(title="Static2API")
|
|
|
|
# Add CORS middleware to allow cross-origin requests
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"], # Allows all origins
|
|
allow_credentials=True,
|
|
allow_methods=["*"], # Allows all methods
|
|
allow_headers=["*"], # Allows all headers
|
|
)
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
return {"message": "Welcome to Static2API"}
|
|
|
|
def load_csv(file_path):
|
|
data = []
|
|
with open(file_path, 'r') as f:
|
|
reader = csv.DictReader(f)
|
|
for row in reader:
|
|
data.append(dict(row))
|
|
return data
|
|
|
|
def load_json(file_path):
|
|
with open(file_path, 'r') as f:
|
|
return json.load(f)
|
|
|
|
@app.on_event("startup")
|
|
async def startup_event():
|
|
# Register API endpoints for all files in the source directory
|
|
register_api_endpoints()
|
|
|
|
def register_api_endpoints():
|
|
source_dir = Path("source")
|
|
if not source_dir.exists():
|
|
return
|
|
|
|
for file_path in source_dir.glob("*"):
|
|
if file_path.is_file():
|
|
endpoint_name = file_path.stem
|
|
route = f"/api/{endpoint_name}"
|
|
src_route = f"/api/{endpoint_name}/src"
|
|
|
|
# Skip if route already exists
|
|
if any(route == route_info.path for route_info in app.routes):
|
|
continue
|
|
|
|
if file_path.suffix.lower() == '.csv':
|
|
# Create a closure to capture the current file_path
|
|
async def get_csv_data(request: Request, file_path=file_path):
|
|
try:
|
|
data = load_csv(file_path)
|
|
return data
|
|
except Exception as e:
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"error": f"Failed to load CSV: {str(e)}"}
|
|
)
|
|
|
|
# Create a closure for serving the original file
|
|
async def get_csv_src(request: Request, file_path=file_path):
|
|
return FileResponse(file_path, media_type='text/csv', filename=file_path.name)
|
|
|
|
# Add the routes
|
|
app.add_api_route(route, get_csv_data, methods=["GET"])
|
|
app.add_api_route(src_route, get_csv_src, methods=["GET"])
|
|
|
|
elif file_path.suffix.lower() == '.json':
|
|
# Create a closure to capture the current file_path
|
|
async def get_json_data(request: Request, file_path=file_path):
|
|
try:
|
|
data = load_json(file_path)
|
|
return data
|
|
except Exception as e:
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"error": f"Failed to load JSON: {str(e)}"}
|
|
)
|
|
|
|
# Create a closure for serving the original file
|
|
async def get_json_src(request: Request, file_path=file_path):
|
|
return FileResponse(file_path, media_type='application/json', filename=file_path.name)
|
|
|
|
# Add the routes
|
|
app.add_api_route(route, get_json_data, methods=["GET"])
|
|
app.add_api_route(src_route, get_json_src, methods=["GET"])
|
|
|
|
@app.middleware("http")
|
|
async def check_for_new_files(request: Request, call_next):
|
|
# Re-register endpoints before processing each request
|
|
# This ensures new files added while the server is running get endpoints
|
|
register_api_endpoints()
|
|
response = await call_next(request)
|
|
return response
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=8000) |