Deploy FastAPI to HF Space

.gitignore

@@ -0,0 +1,4 @@
+venv/
+env/
+ENV/
+.venv/

New folder/.env1

@@ -0,0 +1,2 @@
+QDRANT_URL='https://b6138c60-0a19-4ba7-b6a5-f70a7d653b57.us-west-1-0.aws.cloud.qdrant.io'
+QDRANT_API_KEY="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJhY2Nlc3MiOiJtIn0.XQrkVFAz02zgcvVYbmoneq36biKdbP6491n5I-RrCpQ"

New folder/embedding_service2.py

@@ -0,0 +1,97 @@
+"""
+Embedding Service for generating image embeddings
+"""
+
+import os
+from typing import List, Dict, Any
+from PIL import Image
+import io
+import numpy as np
+import torch
+from transformers import CLIPProcessor, CLIPModel
+
+
+class ImageEmbeddingModel:
+    """Class for generating embeddings from images using CLIP"""
+    
+    def __init__(self, model_name: str = "openai/clip-vit-base-patch32"):
+        """Initialize the CLIP model
+        
+        Args:
+            model_name: Name of the CLIP model to use
+        """
+        self.model_name = model_name
+        self.device = "cuda" if torch.cuda.is_available() else "cpu"
+        self.model = CLIPModel.from_pretrained(model_name).to(self.device)
+        self.processor = CLIPProcessor.from_pretrained(model_name)
+    
+    def generate_embedding(self, image_data: bytes) -> List[float]:
+        """Generate embedding for an image from binary data
+        
+        Args:
+            image_data: Binary image data
+            
+        Returns:
+            Image embedding as a list of floats
+        """
+        # Load image from binary data
+        image = Image.open(io.BytesIO(image_data)).convert("RGB")
+        return self.generate_embedding_from_pil(image)
+    
+    def generate_embedding_from_pil(self, image: Image.Image) -> List[float]:
+        """Generate embedding for a PIL Image
+        
+        Args:
+            image: PIL Image object
+            
+        Returns:
+            Image embedding as a list of floats
+        """
+        # Process image for CLIP
+        inputs = self.processor(images=image, return_tensors="pt").to(self.device)
+        
+        # Generate embedding
+        with torch.no_grad():
+            image_features = self.model.get_image_features(**inputs)
+            
+        # Normalize embedding and convert to list
+        image_embedding = image_features.cpu().numpy()[0]
+        normalized_embedding = image_embedding / np.linalg.norm(image_embedding)
+        return normalized_embedding.tolist()
+    
+    def get_embeddings_from_folder(self, folder_path: str) -> Dict[str, Any]:
+        """Generate embeddings for all images in a folder
+        
+        Args:
+            folder_path: Path to folder containing images
+            
+        Returns:
+            Dictionary mapping filenames to embeddings
+        """
+        results = {}
+        image_extensions = {'.jpg', '.jpeg', '.png', '.bmp', '.gif', '.webp'}
+        
+        # Check if folder exists
+        if not os.path.exists(folder_path):
+            return {"error": f"Folder {folder_path} does not exist"}
+        
+        # Process each image file
+        for filename in os.listdir(folder_path):
+            if os.path.splitext(filename)[1].lower() in image_extensions:
+                try:
+                    file_path = os.path.join(folder_path, filename)
+                    with open(file_path, 'rb') as f:
+                        image_data = f.read()
+                    
+                    embedding = self.generate_embedding(image_data)
+                    results[filename] = {
+                        "embedding": embedding,
+                        "status": "success"
+                    }
+                except Exception as e:
+                    results[filename] = {
+                        "error": str(e),
+                        "status": "failed"
+                    }
+        
+        return results

New folder/vector_db_service2.py

@@ -0,0 +1,154 @@
+"""
+Vector Database Service implementation for Qdrant
+"""
+
+from typing import List, Dict, Any, Optional
+from qdrant_client import QdrantClient
+from qdrant_client.models import PointStruct, VectorParams, Distance, Record
+
+
+
+class VectorDatabaseClient:
+    """Client for interacting with Qdrant vector database"""
+    
+    def __init__(self, url: str, api_key: str, collection_name: str, embedding_size: int):
+        """Initialize Qdrant client and collection settings
+        
+        Args:
+            url: Qdrant server URL
+            api_key: API key for Qdrant
+            collection_name: Name of the collection to use
+            embedding_size: Size of embedding vectors
+        """
+        self.client = QdrantClient(url=url, api_key=api_key)
+        self.collection_name = collection_name
+        self.embedding_size = embedding_size
+    
+    def ensure_collection_exists(self):
+        """Ensure the collection exists, create it if it doesn't"""
+        collections = [c.name for c in self.client.get_collections().collections]
+        
+        if self.collection_name not in collections:
+            self.client.create_collection(
+                collection_name=self.collection_name,
+                vectors_config=VectorParams(
+                    size=self.embedding_size,
+                    distance=Distance.COSINE
+                )
+            )
+            print(f"✅ Collection '{self.collection_name}' created.")
+        else:
+            print(f"ℹ️ Collection '{self.collection_name}' already exists.")
+    
+    def add_embedding(self, id: str, embedding: List[float], filename: str, metadata: Optional[str] = None) -> str:
+        """Add an embedding to the collection
+        
+        Args:
+            id: Unique ID for the point
+            embedding: Vector embedding
+            filename: Original filename
+            metadata: Optional metadata as JSON string
+            
+        Returns:
+            ID of the added point
+        """
+        payload = {"filename": filename}
+        if metadata:
+            payload["metadata"] = metadata
+            
+        self.client.upsert(
+            collection_name=self.collection_name,
+            points=[
+                PointStruct(
+                    id=id,
+                    vector=embedding,
+                    payload=payload
+                )
+            ]
+        )
+        return id
+    
+    def add_embedding_with_payload(self, id: str, embedding: List[float], payload: Dict[str, Any]) -> str:
+        """Add an embedding with a custom payload
+        
+        Args:
+            id: Unique ID for the point
+            embedding: Vector embedding
+            payload: Dictionary of metadata to store
+            
+        Returns:
+            ID of the added point
+        """
+        self.client.upsert(
+            collection_name=self.collection_name,
+            points=[
+                PointStruct(
+                    id=id,
+                    vector=embedding,
+                    payload=payload
+                )
+            ]
+        )
+        return id
+    
+    def search_by_embedding(self, embedding: List[float], limit: int = 5) -> List[Record]:
+        """Search for similar vectors
+        
+        Args:
+            embedding: Query vector
+            limit: Maximum number of results
+            
+        Returns:
+            List of search results
+        """
+        results = self.client.search(
+            collection_name=self.collection_name,
+            query_vector=embedding,
+            limit=limit
+        )
+        return results
+    
+    def search_by_id(self, id: str, limit: int = 1) -> List[Record]:
+        """Search for similar vectors using an existing vector as query
+        
+        Args:
+            id: ID of the existing vector to use as query
+            limit: Maximum number of results
+            
+        Returns:
+            List of search results
+        """
+        # Get the vector by ID
+        vector = self.client.retrieve(
+            collection_name=self.collection_name,
+            ids=[id]
+        )
+        
+        if not vector or len(vector) == 0:
+            return []
+            
+        # Use the vector to search
+        return self.search_by_embedding(vector[0].vector, limit)
+    
+    def delete_embedding(self, id: str) -> bool:
+        """Delete an embedding from the collection
+        
+        Args:
+            id: ID of the embedding to delete
+            
+        Returns:
+            True if deleted, False if not found
+        """
+        self.client.delete(
+            collection_name=self.collection_name,
+            points_selector=[id]
+        )
+        return True
+    
+    def list_collections(self) -> List[str]:
+        """List all collections in the database
+        
+        Returns:
+            List of collection names
+        """
+        return [c.name for c in self.client.get_collections().collections]

README.md

@@ -1,12 +1,97 @@
----
-title: FastAPI
-emoji: 📚
-colorFrom: indigo
-colorTo: purple
-sdk: static
-pinned: false
-license: apache-2.0
-short_description: Fast API with Marqo model
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Image Similarity Search API
+
+A FastAPI application for image similarity search using CLIP embeddings and Qdrant vector database.
+
+## Features
+
+- Upload images and store their vector embeddings
+- Search for similar images using an uploaded image or base64 encoded image
+- Secure API with API key authentication
+- Well-organized, modular codebase following OOP principles
+
+## Installation
+
+1. Clone this repository
+2. Install dependencies:
+
+```bash
+pip install -r requirements.txt
+```
+
+3. Set up environment variables (optional, defaults are provided):
+
+```bash
+export QDRANT_URL="your-qdrant-url"
+export QDRANT_API_KEY="your-qdrant-api-key"
+export COLLECTION_NAME="your-collection-name"
+export API_KEY="your-api-key"
+export PORT=8000
+export ENVIRONMENT="production"  # Or "development" for debug mode with auto-reload
+```
+
+## Usage
+
+Run the application:
+
+```bash
+python app.py
+```
+
+The API will be available at http://localhost:8000 (or the port specified in environment variables).
+
+### API Documentation
+
+Once running, API documentation is available at:
+- Swagger UI: http://localhost:8000/docs
+- ReDoc: http://localhost:8000/redoc
+
+## API Endpoints
+
+- `POST /add-image/`: Add an image to the database
+- `POST /add-images-from-folder/`: Add all images from a folder to the database
+- `POST /search-by-image/`: Search for similar images using an uploaded image
+- `POST /search-by-image-scan/`: Search for similar images using a base64 encoded image
+- `GET /collections`: List all collections in the database
+- `GET /health`: Health check endpoint
+
+## Project Structure
+
+```
+image_similarity_api/
+│
+├── app.py                 # Main application entry point
+├── config.py              # Configuration settings
+├── models/
+│   ├── __init__.py
+│   └── schemas.py         # Pydantic models
+├── services/
+│   ├── __init__.py
+│   ├── embedding.py       # Image embedding service
+│   ├── security.py        # Security service
+│   └── vector_db.py       # Vector database service
+├── api/
+│   ├── __init__.py
+│   └── routes.py          # API routes
+├── requirements.txt       # Project dependencies
+└── README.md              # Project documentation
+```
+
+## Development
+
+For development, set the ENVIRONMENT variable to "development" for auto-reload:
+
+```bash
+export ENVIRONMENT="development"
+python app.py
+```
+
+## Deployment
+
+This application can be deployed to any platform that supports Python applications:
+
+1. Docker
+2. Kubernetes
+3. Cloud platforms (AWS, GCP, Azure, etc.)
+4. Serverless platforms (with appropriate adapters)
+
+Remember to set all required environment variables in your production environment.

api/routes.py

@@ -0,0 +1,109 @@
+"""
+Routes for the Image Similarity Search API
+Contains all endpoints for the application using your original route implementation
+"""
+
+import uuid
+import base64
+import io
+from typing import List, Optional
+from fastapi import APIRouter, FastAPI, File, UploadFile, Form, Query, Path # type: ignore
+from pydantic import BaseModel
+from PIL import Image
+
+from services.embedding_service import ImageEmbeddingModel
+from services.vector_db_service import VectorDatabaseClient
+
+
+class Base64ImageRequest(BaseModel):
+    """Request model for base64 encoded images"""
+    image_data: str
+
+
+def register_routes(
+    app: FastAPI,
+    embedding_model: ImageEmbeddingModel,
+    vector_db: VectorDatabaseClient,
+):
+    """Register all routes with the FastAPI app"""
+
+    @app.api_route("/", methods=["GET", "HEAD"])
+    async def read_root():
+        return {"status": "API running"}
+    
+    @app.post("/add-image/")
+    async def add_image(
+        file: UploadFile = File(...),
+        item_name: str = Form(...),
+        design_name: str = Form(...),
+        item_price: float = Form(...)
+    ):
+        """Upload an image with product details and store its embedding"""
+        # Process the image to get embedding
+        # image_data = await file.read()
+        embedding = await embedding_model.get_embedding_from_upload(file)
+        
+        # Generate a unique ID
+        image_id = str(uuid.uuid4())
+        
+        # Store additional metadata in payload
+        payload = {
+            "filename": file.filename,
+            "item_name": item_name,
+            "design_name": design_name,
+            "item_price": item_price
+        }
+        
+        # Store in vector database
+        vector_db.add_image(image_id, embedding, payload)
+        
+        return {"message": "Image added successfully", "id": image_id}
+    
+    @app.post("/add-images-from-folder/")
+    async def add_images_from_folder(folder_path: str):
+        """Process and add all images from a specified folder"""
+        embeddings = embedding_model.get_embeddings_from_folder(folder_path)
+        return {"embeddings": embeddings}
+    
+    @app.post("/search-by-image/")
+    async def search_by_image(file: UploadFile = File(...)):
+        """Search for similar images by uploading a file"""
+        # Process the image to get embedding
+        # image_data = await file.read()
+        embedding = await embedding_model.get_embedding_from_upload(file)
+        
+        # Search using the embedding
+        results = vector_db.search_by_vector(embedding, limit=1)
+        
+        # return [
+        #     {
+        #         "id": r.id,
+        #         "score": r.score,
+        #         "payload": r.payload
+        #     }
+        #     for r in results
+        # ]
+        return results
+    
+    @app.post("/search-by-image-scan/")
+    async def search_by_image_scan(request: Base64ImageRequest):
+        """Search for similar images using a base64 encoded image"""
+        # Decode base64 image
+        image_data = request.image_data
+        image_bytes = base64.b64decode(image_data.split(',')[1] if ',' in image_data else image_data)
+        
+        # Convert to PIL Image
+        image = Image.open(io.BytesIO(image_bytes)).convert("RGB")
+        
+        # Process image to get embedding
+        embedding = embedding_model.get_embedding_from_pil(image)
+        
+        # Search using the embedding
+        results = vector_db.search_by_vector(embedding, limit=1)
+        
+        return results
+    
+    @app.get("/collections")
+    def list_collections():
+        """List all available collections in the vector database"""
+        return vector_db.list_collections()

api/routes1.py

@@ -0,0 +1,98 @@
+"""
+Routes for the Image Similarity Search API
+Contains all endpoints for the application
+"""
+
+from fastapi import APIRouter, FastAPI, File, UploadFile, Form, Query, Path 
+from typing import List, Optional
+from pydantic import BaseModel
+
+from services.embedding_service import ImageEmbeddingModel
+from services.vector_db_service import VectorDatabaseClient
+
+
+class SearchResponse(BaseModel):
+    """Response model for search results"""
+    image_id: str
+    similarity: float
+    metadata: Optional[dict] = None
+
+
+def register_routes(
+    app: FastAPI,
+    embedding_model: ImageEmbeddingModel,
+    vector_db: VectorDatabaseClient,
+    # Remove security_service parameter
+):
+    """Register all routes with the FastAPI app"""
+    router = APIRouter()
+
+    @router.post("/upload", response_model=dict)
+    async def upload_image(
+        file: UploadFile = File(...),
+        metadata: Optional[str] = Form(None),
+        # Remove security dependency: api_key: str = Depends(security_service.verify_api_key)
+    ):
+        """Upload an image and store its embedding"""
+        # Process the image and generate embedding
+        image_data = await file.read()
+        embedding = embedding_model.generate_embedding(image_data)
+        
+        # Store in vector database with optional metadata
+        image_id = vector_db.add_embedding(embedding, file.filename, metadata)
+        
+        return {"image_id": image_id, "message": "Image uploaded successfully"}
+    
+    @router.get("/search/by-id/{image_id}", response_model=List[SearchResponse])
+    async def search_by_id(
+        image_id: str = Path(..., description="ID of the uploaded image to use as query"),
+        limit: int = Query(5, description="Maximum number of results to return"),
+        # Remove security dependency: api_key: str = Depends(security_service.verify_api_key)
+    ):
+        """Search for similar images using an existing image ID as the query"""
+        results = vector_db.search_by_id(image_id, limit)
+        return [
+            SearchResponse(
+                image_id=result.id,
+                similarity=result.score,
+                metadata=result.metadata
+            )
+            for result in results
+        ]
+    
+    @router.post("/search/by-image", response_model=List[SearchResponse])
+    async def search_by_image(
+        file: UploadFile = File(...),
+        limit: int = Query(5, description="Maximum number of results to return"),
+        # Remove security dependency: api_key: str = Depends(security_service.verify_api_key)
+    ):
+        """Search for similar images by uploading a new image"""
+        # Process the image and generate embedding
+        image_data = await file.read()
+        embedding = embedding_model.generate_embedding(image_data)
+        
+        # Search using the embedding
+        results = vector_db.search_by_embedding(embedding, limit)
+        return [
+            SearchResponse(
+                image_id=result.id,
+                similarity=result.score,
+                metadata=result.metadata
+            )
+            for result in results
+        ]
+    
+    @router.delete("/images/{image_id}")
+    async def delete_image(
+        image_id: str = Path(..., description="ID of the image to delete"),
+        # Remove security dependency: api_key: str = Depends(security_service.verify_api_key)
+    ):
+        """Delete an image from the database"""
+        success = vector_db.delete_embedding(image_id)
+        if success:
+            return {"message": f"Image {image_id} deleted successfully"}
+        return {"message": f"Image {image_id} not found"}
+    
+    
+    # Add the router to the app
+    app.include_router(router, prefix="/api/v1")

app.py

@@ -0,0 +1,125 @@
+"""
+Image Similarity Search API with FastAPI and Qdrant - Fixed Access
+This application provides endpoints for uploading images and searching for similar images
+using vector embeddings from the CLIP model. Implemented using OOP principles.
+"""
+
+import uvicorn # type: ignore
+from fastapi import FastAPI # type: ignore
+from contextlib import asynccontextmanager
+import os
+import ssl
+from fastapi.middleware.cors import CORSMiddleware # type: ignore
+
+from config import Config
+from services.embedding_service import ImageEmbeddingModel
+from services.vector_db_service import VectorDatabaseClient
+from api.routes import register_routes
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """Lifespan context manager for FastAPI application startup and shutdown events"""
+    # This runs before the application starts
+    vector_db = app.state.vector_db
+    vector_db.ensure_collection_exists()
+    
+    yield  # This yields control back to FastAPI
+    
+    # This runs when the application is shutting down
+    # Cleanup code can go here if needed
+
+
+class ImageSimilarityAPI:
+    """Main application class that orchestrates all components"""
+    
+    def __init__(self):
+        # Initialize config
+        self.config = Config()
+        
+        # Initialize components
+        self.embedding_model = ImageEmbeddingModel(self.config.model_name)
+        self.vector_db = VectorDatabaseClient(
+            self.config.qdrant_url, 
+            self.config.qdrant_api_key,
+            self.config.collection_name,
+            self.config.embedding_size
+        )
+        
+        # Initialize FastAPI app with lifespan handler
+        self.app = FastAPI(
+            title="Image Similarity Search API",
+            description="API for uploading images and searching for similar images using CLIP embeddings",
+            version="1.0.0",
+            lifespan=lifespan
+        )
+        
+         # ✅ Enable CORS to allow mobile access
+        self.app.add_middleware(
+            CORSMiddleware,
+            allow_origins=["*"],  # Or set to ["http://192.168.1.42"] for better security
+            allow_credentials=True,
+            allow_methods=["*"],
+            allow_headers=["*"],
+        )
+        # Store vector_db in app state for use in lifespan
+        self.app.state.vector_db = self.vector_db
+        
+        # Register routes
+        register_routes(self.app, self.embedding_model, self.vector_db)
+    
+    def run(self, use_https=False, cert_file="./certs/cert.pem", key_file="./certs/key.pem"):
+        """Run the FastAPI application with optional HTTPS support
+        
+        Args:
+            use_https: Whether to use HTTPS or plain HTTP   
+            cert_file: Path to SSL certificate file
+            key_file: Path to SSL private key file
+        """
+        host = "0.0.0.0"  # Use localhost instead of 0.0.0.0 for better access
+        port = 8000 if not use_https else 8443
+        
+        ssl_context = None
+        if use_https:
+            # Check if certificate files exist
+            if not os.path.exists(cert_file) or not os.path.exists(key_file):
+                print(f"ERROR: SSL certificate files not found at {cert_file} and/or {key_file}")
+                print("Falling back to HTTP. To use HTTPS, please provide valid certificate files.")
+                use_https = False
+            else:
+                # Create SSL context for HTTPS
+                ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
+                ssl_context.load_cert_chain(cert_file, key_file)
+        
+        # Print access URLs for convenience
+        protocol = "https" if use_https else "http"
+        print(f"\n{'='*50}")
+        print(f"Access the API at: {protocol}://{host}:{port}")
+        print(f"Swagger UI available at: {protocol}://{host}:{port}/docs")
+        print(f"ReDoc UI available at: {protocol}://{host}:{port}/redoc")
+        print(f"{'='*50}\n")
+        
+        uvicorn.run(
+            self.app, 
+            host=host,
+            port=port,
+            reload=self.config.environment == "development",
+            ssl_certfile=cert_file if use_https else None,
+            ssl_keyfile=key_file if use_https else None
+        )
+
+
+def create_app() -> FastAPI:
+    """Create and return the FastAPI application"""
+    api = ImageSimilarityAPI()
+    return api.app
+
+
+if __name__ == "__main__":
+    api = ImageSimilarityAPI()
+    # Set to False for now until certificates are properly set up
+    api.run(
+        use_https=False,  # Change to True when certificates are ready
+        cert_file="./certs/cert.pem",
+        key_file="./certs/key.pem"
+    )

config.py

@@ -0,0 +1,21 @@
+"""
+Configuration settings for the Image Similarity API
+"""
+
+import os
+
+
+class Config:
+    """Configuration class for the application"""
+    
+    def __init__(self):
+        self.qdrant_url = os.getenv("QDRANT_URL", 
+            "https://b6138c60-0a19-4ba7-b6a5-f70a7d653b57.us-west-1-0.aws.cloud.qdrant.io")
+        self.qdrant_api_key = os.getenv("QDRANT_API_KEY", 
+            "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJhY2Nlc3MiOiJtIn0.XQrkVFAz02zgcvVYbmoneq36biKdbP6491n5I-RrCpQ")
+        self.collection_name = os.getenv("COLLECTION_NAME", "marqe_embedings")
+        # self.api_key = os.getenv("API_KEY", "your-api-key-here")
+        self.model_name = os.getenv("MODEL_NAME", "hf-hub:Marqo/marqo-ecommerce-embeddings-L")
+        self.embedding_size = 768
+        self.port = int(os.getenv("PORT", 8000))
+        self.environment = os.getenv("ENVIRONMENT", "production")

models/__init__.py

@@ -0,0 +1,5 @@
+"""
+Empty __init__.py file to make the models directory a proper Python package
+"""
+
+# This file is intentionally left empty

models/schemas.py

@@ -0,0 +1,22 @@
+"""
+Pydantic model schemas for API request and response types
+"""
+
+from pydantic import BaseModel
+
+
+class Base64ImageRequest(BaseModel):
+    """Model for accepting base64 encoded images"""
+    image_data: str
+
+
+class SearchResult(BaseModel):
+    """Model for search results"""
+    id: str
+    score: float
+    payload: dict
+
+
+class ErrorResponse(BaseModel):
+    """Model for error responses"""
+    detail: str

requirements.txt

@@ -0,0 +1,13 @@
+# requirements.txt
+torch
+open_clip_torch
+fastapi 
+uvicorn
+qdrant-client 
+nest_asyncio
+python-multipart
+pillow
+numpy
+pydantic
+python-dotenv
+

services/__init__.py

@@ -0,0 +1,5 @@
+"""
+Empty __init__.py file to make the api directory a proper Python package
+"""
+
+# This file is intentionally left empty

services/embedding_service.py

@@ -0,0 +1,88 @@
+"""
+Image embedding service for generating vector embeddings from images
+"""
+
+import io
+import os
+import base64
+from typing import List, Tuple
+
+import open_clip
+import torch
+from fastapi import UploadFile, HTTPException
+from PIL import Image
+import torch
+
+
+class ImageEmbeddingModel:
+    """Class for handling image embedding using CLIP model"""
+    
+    def __init__(self, model_name: str):
+        self.model_name = model_name
+        self.device = "cuda" if torch.cuda.is_available() else "cpu"
+        self.model, self.preprocess_train, self.preprocess_val = self._initialize_model()
+        
+    def _initialize_model(self) -> Tuple:
+        """Initialize the CLIP model for image embeddings"""
+        model, preprocess_train, preprocess_val = open_clip.create_model_and_transforms(self.model_name)
+        tokenizer = open_clip.get_tokenizer(self.model_name)
+        model.to(self.device)
+        model.eval()
+        return model, preprocess_train, preprocess_val
+    
+    def get_embedding_from_pil(self, image: Image.Image) -> List[float]:
+        """Get embedding from PIL image"""
+        processed_image = self.preprocess_val(image).unsqueeze(0).to(self.device)
+
+        # with torch.no_grad(), torch.amp.autocast(device_type='cuda' if self.device == 'cuda' else 'cpu')
+        if self.device == 'cuda':
+            autocast_context = torch.amp.autocast(device_type='cuda')
+        else:
+            # On CPU, autocast should either be skipped or forced to float32
+            autocast_context = torch.amp.autocast(device_type='cpu', dtype=torch.float32)
+        with torch.no_grad(), autocast_context:
+            image_features = self.model.encode_image(processed_image, normalize=True)
+
+
+        return image_features.cpu().numpy()[0].tolist()
+    
+    async def get_embedding_from_upload(self, image_file: UploadFile) -> List[float]:
+        """Get embedding from uploaded image file"""
+        try:
+            contents = await image_file.read()
+            img = Image.open(io.BytesIO(contents)).convert("RGB")
+            return self.get_embedding_from_pil(img)
+        except Exception as e:
+            raise HTTPException(status_code=400, detail=f"Invalid image: {str(e)}")
+    
+    def get_embedding_from_base64(self, base64_data: str) -> List[float]:
+        """Get embedding from base64 encoded image"""
+        try:
+            # Handle data URI format
+            if ',' in base64_data:
+                base64_data = base64_data.split(',')[1]
+                
+            image_bytes = base64.b64decode(base64_data)
+            image = Image.open(io.BytesIO(image_bytes)).convert("RGB")
+            
+            return self.get_embedding_from_pil(image)
+        except Exception as e:
+            raise HTTPException(status_code=400, detail=f"Invalid base64 image: {str(e)}")
+    
+    def get_embeddings_from_folder(self, image_folder: str) -> List[List[float]]:
+        """Get embeddings from all images in a folder"""
+        embeddings = []
+
+        if not os.path.exists(image_folder):
+            raise HTTPException(status_code=404, detail=f"Folder not found: {image_folder}")
+
+        for image_name in os.listdir(image_folder):
+            if image_name.lower().endswith(('.png', '.jpg', '.jpeg')):
+                try:
+                    image_path = os.path.join(image_folder, image_name)
+                    img = Image.open(image_path).convert("RGB")
+                    embeddings.append(self.get_embedding_from_pil(img))
+                except Exception as e:
+                    print(f"Error processing {image_name}: {str(e)}")
+
+        return embeddings

services/security_service.py

@@ -0,0 +1,20 @@
+"""
+Security service for API authentication and authorization
+"""
+
+from fastapi import Depends, HTTPException
+from fastapi.security import APIKeyHeader
+
+
+class SecurityService:
+    """Class for handling API security"""
+    
+    def __init__(self, api_key: str):
+        self.api_key = api_key
+        self.api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
+    
+    async def verify_api_key(self, api_key: str = Depends(APIKeyHeader(name="X-API-Key", auto_error=False))):
+        """Verify API key dependency"""
+        if self.api_key != "your-api-key-here" and api_key != self.api_key:  # Skip check if using default key
+            raise HTTPException(status_code=401, detail="Invalid API key")
+        return api_key

services/vector_db_service.py

@@ -0,0 +1,70 @@
+"""
+Vector database service for interacting with Qdrant
+"""
+
+from typing import List, Dict, Any
+
+from fastapi import HTTPException # type: ignore
+from qdrant_client import QdrantClient # type: ignore
+from qdrant_client.models import Distance, PointStruct, VectorParams # type: ignore
+
+class VectorDatabaseClient:
+    """Class for interacting with Qdrant vector database"""
+    
+    def __init__(self, url: str, api_key: str, collection_name: str, embedding_size: int):
+        self.url = url
+        self.api_key = api_key
+        self.collection_name = collection_name
+        self.embedding_size = embedding_size
+        self.client = QdrantClient(url=url, api_key=api_key)
+        
+    def ensure_collection_exists(self) -> None:
+        """Ensure the Qdrant collection exists"""
+        collections = self.client.get_collections()
+        collection_names = [c.name for c in collections.collections]
+        
+        if self.collection_name not in collection_names:
+            self.client.create_collection(
+                collection_name=self.collection_name,
+                vectors_config=VectorParams(
+                    size=self.embedding_size,
+                    distance=Distance.COSINE
+                )
+            )
+            print(f"✅ Collection '{self.collection_name}' created.")
+        else:
+            print(f"ℹ️ Collection '{self.collection_name}' already exists.")
+    
+    def add_image(self, image_id: str, embedding: List[float], payload: Dict[str, Any]) -> None:
+        """Add an image embedding to the database"""
+        self.client.upsert(
+            collection_name=self.collection_name,
+            points=[
+                PointStruct(
+                    id=image_id,
+                    vector=embedding,
+                    payload=payload
+                )
+            ]
+        )
+    
+    def search_by_vector(self, embedding: List[float], limit: int = 1) -> List[Dict[str, Any]]:
+        """Search for similar images using an embedding vector"""
+        results = self.client.search(
+            collection_name=self.collection_name,
+            query_vector=embedding,
+            limit=limit
+        )
+        
+        return [
+            {
+                "id": r.id,
+                "score": r.score,
+                "payload": r.payload
+            }
+            for r in results
+        ]
+    
+    def list_collections(self) -> List[str]:
+        """List all collections in the database"""
+        return [c.name for c in self.client.get_collections().collections]