initial project

.dockerignore

@@ -0,0 +1,2 @@
+__pycache__
+.env

.gitignore

@@ -0,0 +1,2 @@
+__pycache__
+.env

Dockerfile

@@ -0,0 +1,14 @@
+# Use the official Python 3.10.9 image
+FROM python:3.10.9
+
+# Copy the current directory contents into the container at .
+COPY . .
+
+# Set the working directory to /
+WORKDIR /
+
+# Install requirements.txt 
+RUN pip install --no-cache-dir --upgrade -r /requirements.txt
+
+# Start the FastAPI app on port 7860, the default port expected by Spaces
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,9 +1,11 @@
 ---
-title: Qccamapi
-emoji: 👁
-colorFrom: purple
-colorTo: indigo
-sdk: docker
+title: QCCam APIs
+emoji: 📈
+colorFrom: red
+colorTo: green
+sdk: gradio
+sdk_version: 4.44.0
+app_file: app.py
 pinned: false
 ---
 

apis/s3.py

@@ -0,0 +1,50 @@
+import os
+import boto3
+from dotenv import load_dotenv
+load_dotenv()
+
+class S3:
+
+    def __init__(self) -> None:
+        self.aws_access_key_id = os.getenv('AWS_ACCESS_KEY_ID', '')
+        self.aws_secret_access_key = os.getenv('AWS_SECRET_ACCESS_KEY', '')
+        self.aws_default_region = os.getenv('AWS_DEFAULT_REGION', '')
+
+        self.client = boto3.client(
+            service_name='s3',
+            region_name=self.aws_default_region,
+            aws_access_key_id=self.aws_access_key_id,
+            aws_secret_access_key=self.aws_secret_access_key
+        )
+
+    # Function to list images in a specific S3 bucket
+    def list_images_in_bucket(self, bucket_name):
+        response = self.client.list_objects_v2(Bucket=bucket_name)
+        return [obj['Key'] for obj in response.get('Contents', []) if obj['Key'].lower().endswith(('.png', '.jpg', '.jpeg'))]
+    
+    # Function to get image URLs
+    def get_image_urls(self, bucket_name, image_keys):
+        return [self.client.generate_presigned_url('get_object', Params={'Bucket': bucket_name, 'Key': key}) for key in image_keys]
+
+    # Function to upload file to S3
+    def upload_file(self, bucket_name, file, name):
+        print(f'Uploading "{name}"...')
+        return self.client.upload_fileobj(file, bucket_name, name)
+
+    # Function to delete a file from S3
+    def delete_image(self, bucket_name, file_key):
+        if isinstance(file_key, list):
+            for k in file_key:
+                print(f'Removing "{k}"...')
+                self.client.delete_object(Bucket=bucket_name, Key=k)
+        else:
+            self.client.delete_object(Bucket=bucket_name, Key=file_key)
+
+    # Function to list images in a specific S3 bucket
+    def list_files_in_bucket(self, bucket_name):
+        response = self.client.list_objects_v2(Bucket=bucket_name)
+        return [obj['Key'] for obj in response.get('Contents', [])]
+
+    # Function to get image URLs
+    def get_files_urls(self, bucket_name, file_keys):
+        return [self.client.generate_presigned_url('get_object', Params={'Bucket': bucket_name, 'Key': key}) for key in file_keys]

app.py

@@ -0,0 +1,76 @@
+import os
+from typing import Annotated
+from fastapi import FastAPI, Header
+from apis.s3 import S3
+from dotenv import load_dotenv
+from models.image_object import ImageObject
+from models.common_headers import CommonHeaders
+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', '')
+USERNAME = os.environ.get('USERNAME')
+PASSWORD = os.environ.get('PASSWORD')
+
+s3client = S3()
+app = FastAPI(docs_url=None, redoc_url=None)
+
+@app.get("/docs", include_in_schema=False)
+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",
+    )
+
+
+@app.get(app.swagger_ui_oauth2_redirect_url, include_in_schema=False)
+async def swagger_ui_redirect():
+    return get_swagger_ui_oauth2_redirect_html()
+
+@app.get("/redoc", include_in_schema=False)
+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",
+    )
+
+@app.get("/")
+def root():
+    return {"message": "ok"}
+        
+@app.get("/health")
+def healthcheck():
+    return {"message": "ok"}
+        
+@app.post("/image")
+def upload(
+    o: ImageObject,
+    headers: Annotated[CommonHeaders, Header()]
+):
+    res = None
+    if headers.x_request_user == USERNAME \
+       and headers.x_api_key == PASSWORD:
+        key = f'{o.key}/{o.job_no}/{o.name}'
+        res = s3client.upload_file(AWS_S3_BUCKET_NAME, o.content, key)
+    return res
+
+# # Create a Gradio interface
+# with gr.Blocks() as demo:
+#     pass
+   
+# def auth(username, password):
+#     u = os.environ.get('USERNAME')
+#     p = os.environ.get('PASSWORD')
+#     return (username == u and password == p)
+ 
+# # Launch the Gradio app
+# demo.launch(auth=auth if not IS_DEV else None)

models/common_headers.py

@@ -0,0 +1,6 @@
+from pydantic import BaseModel
+from typing import Union
+
+class CommonHeaders(BaseModel):
+    x_request_user: str
+    x_api_key: str

models/image_object.py

@@ -0,0 +1,8 @@
+from pydantic import BaseModel
+from typing import Union
+
+class ImageObject(BaseModel):
+    name: str
+    key: str
+    job_no: str
+    content: Union[str, None] = None

requirements.txt

@@ -0,0 +1,7 @@
+boto3
+gradio
+pillow
+pydantic
+python-dotenv
+fastapi[standard]
+uvicorn