Spaces:
Sleeping
Sleeping
File size: 1,992 Bytes
28bbb93 ffce8b3 0b89ddd 28bbb93 d94a30f 28bbb93 0b89ddd 28bbb93 9087412 28bbb93 008aefe 28bbb93 d94a30f 28bbb93 ffc41ed 28bbb93 ffce8b3 28bbb93 18367a2 28bbb93 ffce8b3 28bbb93 1ec7cc7 28bbb93 |
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 |
import asyncio
from concurrent.futures import ThreadPoolExecutor
from fastapi import FastAPI, Response, Request, BackgroundTasks
from starlette.middleware.cors import CORSMiddleware
# import json
# from load_data import get_design_data, get_design_resolutions_for_shop
from starlette.responses import RedirectResponse
from createlookalike import create_feature_files
from schemas import Shop
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Allows all origins
allow_credentials=True,
allow_methods=["*"], # Allows all methods
allow_headers=["*"], # Allows all headers
)
API_TOKEN = '34dsadfFTGREGEFGE'
# ---------------------------------------------------------------------------
# - Token auth method -
# ---------------------------------------------------------------------------
def check_auth(authorization_header):
if authorization_header == API_TOKEN:
return True
else:
return False
@app.get("/")
async def root():
# return {"message": "where are you looking for?"}
response = RedirectResponse(url='/docs')
return response
@app.get("/active")
async def active(request: Request, response: Response):
authorization_token = request.headers.get("Authorization", None)
check_token = check_auth(authorization_token)
return {"application":"active"}
@app.post("/lal_for_shop")
async def submit_async_task(request: Request, response: Response,background_tasks: BackgroundTasks, shop: Shop):
# loop = asyncio.new_event_loop()
authorization_token = request.headers.get("Authorization", None)
check_token = check_auth(authorization_token)
if check_token:
# task = asyncio.create_task(create_feature_files(shop))
background_tasks.add_task(create_feature_files, shop)
return {"message": f"shop calculation for shop {shop.id} submitted"}
else:
print(authorization_token)
response.status_code = 401
return response
|