Spaces:
Sleeping
Sleeping
File size: 3,310 Bytes
192f79a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 |
import os
import json
from fastapi import FastAPI, HTTPException, Depends, Request
from fastapi.responses import JSONResponse
from fastapi.security.api_key import APIKeyHeader
from dotenv import load_dotenv
from typing import List
# Load environment variables
load_dotenv()
API_KEY = os.getenv("API_KEY")
API_KEY_NAME = "access_token"
# Debug print
print(f"[DEBUG] Loaded API_KEY: {API_KEY}")
# Ensure data directory exists
DATA_DIR = "data"
os.makedirs(DATA_DIR, exist_ok=True)
print(f"[DEBUG] Ensured data directory exists at: {DATA_DIR}")
# API key dependency
api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=False)
def get_api_key(api_key_header: str = Depends(api_key_header)):
print(f"[DEBUG] Checking API key: {api_key_header}")
if api_key_header == API_KEY:
return api_key_header
else:
print("[DEBUG] Invalid API key!")
raise HTTPException(status_code=403, detail="Could not validate credentials")
app = FastAPI()
@app.post("/files/", dependencies=[Depends(get_api_key)])
def create_json_file(filename: str, content: dict):
print(f"[DEBUG] Creating file: {filename}.json with content: {content}")
file_path = os.path.join(DATA_DIR, f"{filename}.json")
if os.path.exists(file_path):
print("[DEBUG] File already exists!")
raise HTTPException(status_code=400, detail="File already exists")
with open(file_path, "w", encoding="utf-8") as f:
json.dump(content, f, indent=2)
return {"message": "File created", "filename": filename}
@app.get("/files/", dependencies=[Depends(get_api_key)])
def list_json_files() -> List[str]:
print("[DEBUG] Listing all JSON files")
files = [f for f in os.listdir(DATA_DIR) if f.endswith('.json')]
return files
@app.get("/files/{filename}", dependencies=[Depends(get_api_key)])
def read_json_file(filename: str):
print(f"[DEBUG] Reading file: {filename}.json")
file_path = os.path.join(DATA_DIR, f"{filename}.json")
if not os.path.exists(file_path):
print("[DEBUG] File not found!")
raise HTTPException(status_code=404, detail="File not found")
with open(file_path, "r", encoding="utf-8") as f:
data = json.load(f)
return data
@app.put("/files/{filename}", dependencies=[Depends(get_api_key)])
def edit_json_file(filename: str, content: dict):
print(f"[DEBUG] Editing file: {filename}.json with content: {content}")
file_path = os.path.join(DATA_DIR, f"{filename}.json")
if not os.path.exists(file_path):
print("[DEBUG] File not found!")
raise HTTPException(status_code=404, detail="File not found")
with open(file_path, "w", encoding="utf-8") as f:
json.dump(content, f, indent=2)
return {"message": "File updated", "filename": filename}
@app.delete("/files/{filename}", dependencies=[Depends(get_api_key)])
def delete_json_file(filename: str):
print(f"[DEBUG] Deleting file: {filename}.json")
file_path = os.path.join(DATA_DIR, f"{filename}.json")
if not os.path.exists(file_path):
print("[DEBUG] File not found!")
raise HTTPException(status_code=404, detail="File not found")
os.remove(file_path)
return {"message": "File deleted", "filename": filename} |