implement backend API

Author: ks6088ts

azure_ai_vision.env.sample

@@ -0,0 +1,2 @@
+AZURE_AI_VISION_API_KEY="<your-api-key>"
+AZURE_AI_VISION_ENDPOINT="<your-endpoint>"

backend/fastapi.py

@@ -1,6 +1,7 @@
 from fastapi import FastAPI
 from fastapi.openapi.utils import get_openapi
 
+from backend.routers import azure_ai_vision as azure_ai_vision_router
 from backend.routers import azure_openai as azure_openai_router
 from backend.routers import azure_storage as azure_storage_router
 from backend.routers import document_intelligence as document_intelligence_router
@@ -12,6 +13,7 @@
 app.include_router(azure_openai_router.router)
 app.include_router(document_intelligence_router.router)
 app.include_router(azure_storage_router.router)
+app.include_router(azure_ai_vision_router.router)
 
 
 def custom_openapi():

backend/internals/azure_ai_vision.py

@@ -0,0 +1,35 @@
+from logging import getLogger
+
+from azure.ai.vision.imageanalysis import ImageAnalysisClient
+from azure.ai.vision.imageanalysis.models import VisualFeatures
+from azure.core.credentials import AzureKeyCredential
+
+from backend.settings.azure_ai_vision import Settings
+
+logger = getLogger(__name__)
+
+
+class AzureAiVisionClient:
+    def __init__(self, settings: Settings) -> None:
+        self.settings = settings
+
+    def get_image_analysis_client(self) -> ImageAnalysisClient:
+        return ImageAnalysisClient(
+            endpoint=self.settings.azure_ai_vision_endpoint,
+            credential=AzureKeyCredential(self.settings.azure_ai_vision_api_key),
+        )
+
+    def analyze_image(
+        self,
+        image: bytes,
+    ) -> dict:
+        image_analysis_client = self.get_image_analysis_client()
+        result = image_analysis_client.analyze(
+            image_data=image,
+            visual_features=[
+                VisualFeatures.CAPTION,
+                VisualFeatures.READ,
+            ],
+        )
+        logger.info("Analyzed image")
+        return result.as_dict()

backend/routers/azure_ai_vision.py

@@ -0,0 +1,37 @@
+from logging import getLogger
+
+from fastapi import APIRouter, UploadFile
+
+from backend.internals import azure_ai_vision
+from backend.schemas import azure_ai_vision as azure_ai_vision_schemas
+from backend.settings.azure_ai_vision import Settings
+
+logger = getLogger(__name__)
+client = azure_ai_vision.AzureAiVisionClient(
+    settings=Settings(),
+)
+
+router = APIRouter(
+    prefix="/azure_ai_vision",
+    tags=["azure_ai_vision"],
+    responses={404: {"description": "Not found"}},
+)
+
+
+@router.post(
+    "/image/analyze/",
+    response_model=azure_ai_vision_schemas.ImageAnalysisResponse,
+    status_code=200,
+)
+async def analyze_image(file: UploadFile):
+    try:
+        content = await file.read()
+        result = client.analyze_image(
+            image=content,
+        )
+    except Exception as e:
+        logger.error(f"Failed to analyze image: {e}")
+        raise
+    return azure_ai_vision_schemas.ImageAnalysisResponse(
+        result=result,
+    )

backend/schemas/azure_ai_vision.py

@@ -0,0 +1,9 @@
+from logging import getLogger
+
+from pydantic import BaseModel
+
+logger = getLogger(__name__)
+
+
+class ImageAnalysisResponse(BaseModel):
+    result: dict

backend/settings/azure_ai_vision.py

@@ -0,0 +1,11 @@
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    azure_ai_vision_endpoint: str = "https://<name>.cognitiveservices.azure.com/"
+    azure_ai_vision_api_key: str = "<api-key>"
+
+    model_config = SettingsConfigDict(
+        env_file="azure_ai_vision.env",
+        env_file_encoding="utf-8",
+    )