Spaces:
Runtime error
Runtime error
File size: 2,791 Bytes
c6017a8 9bced77 c6017a8 8c73651 c6017a8 8c73651 c6017a8 9bced77 c6017a8 |
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 |
import requests
import uvicorn
from fastapi import FastAPI, Body
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import Response
from pydantic import BaseModel
from tclogger import logger
from typing import Optional
from apis.arg_parser import ArgParser
from configs.envs import DATA_PROXY_API_ENVS, CACHE_ROOT
class DataProxyAPI:
REQUESTS_HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36"
}
def __init__(self, app_envs: dict = {}):
self.title = app_envs.get("app_name")
self.version = app_envs.get("version")
self.app = FastAPI(
docs_url="/",
title=self.title,
version=self.version,
swagger_ui_parameters={"defaultModelsExpandDepth": -1},
)
# self.allow_cors()
self.setup_routes()
logger.success(f"> {self.title} - v{self.version}")
def allow_cors(self):
self.app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
def get_img_path_by_url(self, url: str):
return CACHE_ROOT / url.split("/")[-1]
def save_img(self, img_url: str, img_bytes: bytes):
img_path = self.get_img_path_by_url(img_url)
if not img_path.parent.exists():
img_path.parent.mkdir(parents=True, exist_ok=True)
with open(img_path, "wb") as img_file:
img_file.write(img_bytes)
return img_path
def get_img_bytes_from_url(self, url: str):
try:
response = requests.get(url, headers=self.REQUESTS_HEADERS)
img_bytes = response.content
except Exception as e:
img_bytes = None
return img_bytes
def get_img(self, url: str):
img_path = self.get_img_path_by_url(url)
if img_path.exists():
with open(img_path, "rb") as img_file:
img_bytes = img_file.read()
else:
img_bytes = self.get_img_bytes_from_url(url)
self.save_img(url, img_bytes)
return Response(content=img_bytes, media_type="image/jpeg")
def setup_routes(self):
self.app.get(
"/img",
summary="Get image by url",
)(self.get_img)
if __name__ == "__main__":
app_envs = DATA_PROXY_API_ENVS
app = DataProxyAPI(app_envs).app
app_args = ArgParser(app_envs).args
if app_args.reload:
uvicorn.run("__main__:app", host=app_args.host, port=app_args.port, reload=True)
else:
uvicorn.run("__main__:app", host=app_args.host, port=app_args.port)
# python -m apps.data_proxy_api
|