Spaces:

ravi19
/

FastAPI

No application file

Deploy FastAPI to HF Space

b36cb8b over 1 year ago

4.62 kB

"""
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"
    )

	"""
	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"
	)