File size: 2,441 Bytes
d359c5d |
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 83 84 85 86 87 88 89 90 91 92 93 |
import os
from fastapi import FastAPI, Header, HTTPException
from fastapi.middleware.cors import CORSMiddleware
import boto3
# -----------------------------
# CONFIG
# -----------------------------
DO_SPACES_REGION = "blr1"
DO_SPACES_BUCKET = "milestone"
DO_ACCESS_KEY = os.getenv("DO_SPACES_KEY")
DO_SECRET_KEY = os.getenv("DO_SPACES_SECRET")
BASE_URL = f"https://{DO_SPACES_BUCKET}.{DO_SPACES_REGION}.digitaloceanspaces.com"
FOLDER_PREFIX = "bikini-theme/target/"
AUTH_TOKEN = "logicgo_admin@123"
# -----------------------------
# FASTAPI INIT
# -----------------------------
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# -----------------------------
# SPACES CLIENT
# -----------------------------
def get_spaces_client():
endpoint = f"{DO_SPACES_REGION}.digitaloceanspaces.com"
return boto3.client(
"s3",
region_name=DO_SPACES_REGION,
endpoint_url=f"https://{endpoint}",
aws_access_key_id=DO_ACCESS_KEY,
aws_secret_access_key=DO_SECRET_KEY,
)
# -----------------------------
# AUTH MIDDLEWARE
# -----------------------------
def check_auth(authorization: str = Header(None)):
if not authorization:
raise HTTPException(status_code=401, detail="Missing Authorization header")
token = authorization.replace("Bearer ", "").strip()
if token != AUTH_TOKEN:
raise HTTPException(status_code=403, detail="Invalid authorization token")
# -----------------------------
# PROTECTED ROUTE
# -----------------------------
@app.get("/bikini-theme")
def list_bikini_theme_images(authorization: str = Header(None)):
check_auth(authorization)
client = get_spaces_client()
# List objects
resp = client.list_objects_v2(
Bucket=DO_SPACES_BUCKET,
Prefix=FOLDER_PREFIX
)
contents = resp.get("Contents", [])
image_files = [
obj["Key"]
for obj in contents
if obj["Key"].lower().endswith((".png", ".jpg", ".jpeg", ".webp"))
]
images = {}
for file_key in image_files:
file_name = file_key.split("/")[-1]
name_without_ext = file_name.rsplit(".", 1)[0]
images[name_without_ext] = f"{BASE_URL}/{file_key}"
return {"images": images}
@app.get("/")
def home():
return {"message": "FastAPI DO Spaces bikini-theme API running on HuggingFace"}
|