Spaces:
Build error
Build error
| from http import HTTPStatus | |
| import os, io, base64, json | |
| from typing import Any, Dict, Optional | |
| from fastapi.responses import JSONResponse | |
| from typing_extensions import Annotated | |
| import uvicorn | |
| import logging | |
| from fastapi import FastAPI, File, Form, HTTPException, Header, UploadFile | |
| from apis.s3 import S3 | |
| from dotenv import load_dotenv | |
| from models.response import Response | |
| from models.image_object import ImageObject | |
| from apis.common import Common | |
| from fastapi.openapi.docs import ( | |
| get_redoc_html, | |
| get_swagger_ui_html, | |
| get_swagger_ui_oauth2_redirect_html, | |
| ) | |
| load_dotenv() | |
| IS_DEV = os.environ.get('ENV', 'DEV') != 'PROD' | |
| AWS_S3_BUCKET_NAME = os.getenv('AWS_S3_BUCKET_NAME', '') | |
| X_API_KEY = os.environ.get('X_API_KEY') | |
| logging.basicConfig( | |
| level=logging.DEBUG if IS_DEV else logging.INFO, | |
| format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', | |
| datefmt='%Y-%m-%d %H:%M:%S' | |
| ) | |
| s3client = S3(logging) | |
| common = Common() | |
| app = FastAPI(docs_url=None, redoc_url=None) | |
| async def custom_swagger_ui_html(): | |
| return get_swagger_ui_html( | |
| openapi_url=app.openapi_url, | |
| title=app.title + " - Swagger UI", | |
| oauth2_redirect_url=app.swagger_ui_oauth2_redirect_url, | |
| swagger_js_url="https://unpkg.com/swagger-ui-dist@5/swagger-ui-bundle.js", | |
| swagger_css_url="https://unpkg.com/swagger-ui-dist@5/swagger-ui.css", | |
| ) | |
| async def swagger_ui_redirect(): | |
| return get_swagger_ui_oauth2_redirect_html() | |
| async def redoc_html(): | |
| return get_redoc_html( | |
| openapi_url=app.openapi_url, | |
| title=app.title + " - ReDoc", | |
| redoc_js_url="https://unpkg.com/redoc@next/bundles/redoc.standalone.js", | |
| ) | |
| def root(): | |
| return Response(status=200, data='ok') | |
| def healthcheck(): | |
| return Response(status=200, data='ok') | |
| async def upload( | |
| o: ImageObject, | |
| x_request_user: str = Header(...), | |
| x_api_key: str = Header(...), | |
| metadata: str = Header(...), | |
| ): | |
| res = Response() | |
| logging.info("--------------------------------") | |
| logging.info(f"Received request to upload image by {x_request_user}") | |
| if is_valid(x_api_key): | |
| key = f'{o.key}/{o.job_no}/{o.name}' | |
| logging.info(f'Key for S3 upload: {key}') | |
| if o.content is not None: | |
| try: | |
| metadata = json.loads(metadata) | |
| except Exception as e: | |
| metadata = {} | |
| logging.info(f'Error: {str(e)}\nUse default metadata instead.') | |
| try: | |
| # Decode base64 content | |
| decoded_content = base64.b64decode(o.content) | |
| logging.info(f'Decoded content length: {len(decoded_content)} bytes') | |
| # Wrap bytes in BytesIO to create a file-like object | |
| file_obj = io.BytesIO(decoded_content) | |
| # Upload file object to S3 | |
| # logging.info(f"Uploading file to S3 bucket '{AWS_S3_BUCKET_NAME}' with key '{key}'") | |
| metadata['x-request-id'] = x_request_user | |
| res = s3client.upload_file(AWS_S3_BUCKET_NAME, file_obj, key, metadata=metadata) | |
| logging.info("File uploaded successfully") | |
| except Exception as e: | |
| res.error = str(e) | |
| logging.warning(f"Error during upload: {res.error}") | |
| res.data = o.ref | |
| return res.json() | |
| async def upload2( | |
| x_request_user: str = Header(...), | |
| x_api_key: str = Header(...), | |
| job_no: Annotated[str, Form()] = '', | |
| s3key: Annotated[str, Form()] = '', | |
| ref: Annotated[str, Form()] = '', | |
| fileb: Optional[UploadFile] = File(None), | |
| ): | |
| res = Response() | |
| res.data = ref | |
| logging.info("--------------------------------") | |
| logging.info("Received request to upload image") | |
| # Validate headers | |
| if not is_valid(x_api_key): | |
| res.status = HTTPStatus.FORBIDDEN | |
| res.error = "Invalid credentials" | |
| return res.json() | |
| if fileb is None: | |
| res.status = HTTPStatus.BAD_REQUEST | |
| res.error = "File not found" | |
| return res.json() | |
| key = f'{s3key}/{job_no}/{fileb.filename}' | |
| logging.info(f'Key for S3 upload: {key}') | |
| try: | |
| # Read the file content directly from UploadFile | |
| file_content = await fileb.read() | |
| logging.info(f'File content length: {len(file_content)} bytes') | |
| # # Upload file object to S3 | |
| # # logging.info(f"Uploading file to S3 bucket '{AWS_S3_BUCKET_NAME}' with key '{key}'") | |
| # res = s3client.upload_file(AWS_S3_BUCKET_NAME, file, key) | |
| # logging.info("File uploaded successfully") | |
| except Exception as e: | |
| res.error = str(e) | |
| logging.warning(f"Error during upload: {res.error}") | |
| return res.json() | |
| def is_valid(p): | |
| return p == X_API_KEY | |
| if __name__=='__main__': | |
| uvicorn.run('app:app', host='0.0.0.0', port=7860, reload=True) |