johnpccd/image_management_api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

cp

Browse Source
This commit is contained in:
johnpccd 2025-05-24 11:51:36 +02:00
parent 33c44dcda9
commit d67d5d7e05
3 changed files with 680 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

239
README.md
View File

@ -19,7 +19,8 @@ sereact/
├── images/ # Sample images for testing
├── sereact/ # Main application code
│ ├── deployment/ # Deployment configurations
│ │ └── cloud-run/ # Google Cloud Run configuration
│ │ ├── cloud-run/ # Google Cloud Run configuration
│ │ └── terraform/ # Infrastructure as code
│ ├── docs/ # Documentation
│ │ └── api/ # API documentation
│ ├── scripts/ # Utility scripts
@ -29,6 +30,7 @@ sereact/
│ │ ├── core/ # Core modules
│ │ ├── db/ # Database models and repositories
│ │ │ ├── models/ # Data models
│ │ │ ├── providers/ # Database providers
│ │ │ └── repositories/ # Database operations
│ │ ├── schemas/ # API schemas (request/response)
│ │ └── services/ # Business logic services
@ -41,14 +43,59 @@ sereact/
└── README.md # This file
```
## System Architecture
```
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ │ │ │ │ │
│ FastAPI │ ───────▶│ Firestore │◀────────│ Cloud │
│ Backend │ │ Database │ │ Functions │
│ │ │ │ │ │
└─────┬───────┘ └─────────────┘ └──────┬──────┘
│ │
│ │
▼ │
┌─────────────┐ ┌─────────────┐ │
│ │ │ │ │
│ Cloud │ │ Pub/Sub │ │
│ Storage │────────▶│ Queue │────────────────┘
│ │ │ │
└─────────────┘ └─────────────┘
│
│
▼
┌─────────────┐ ┌─────────────┐
│ │ │ │
│ Cloud │ │ Pinecone │
│ Vision API │────────▶│ Vector DB │
│ │ │ │
└─────────────┘ └─────────────┘
```
1. **Image Upload Flow**:
- Images are uploaded through the FastAPI backend
- Images are stored in Cloud Storage
- A message is published to Pub/Sub queue
2. **Embedding Generation Flow**:
- Cloud Function is triggered by Pub/Sub message
- Function calls Cloud Vision API to generate image embeddings
- Embeddings are stored in Pinecone Vector DB
3. **Search Flow**:
- Search queries processed by FastAPI backend
- Vector similarity search performed against Pinecone
- Results combined with metadata from Firestore
## Technology Stack
- FastAPI - Web framework
- Firestore - Database
- Google Cloud Storage - Image storage
- Google Pub/Sub - Message queue
- Google Cloud Functions - Serverless computing
- Google Cloud Vision API - Image analysis and embedding generation
- Pinecone - Vector database for semantic search
- CLIP - Image embedding model
- NumPy - Scientific computing
- Pydantic - Data validation
## Setup and Installation
@ -56,8 +103,8 @@ sereact/
### Prerequisites
- Python 3.8+
- Google Cloud account with Firestore and Storage enabled
- (Optional) Pinecone account for semantic search
- Google Cloud account with Firestore, Storage, Pub/Sub, and Cloud Functions enabled
- Pinecone account for vector database
### Installation
@ -89,10 +136,16 @@ sereact/
GCS_BUCKET_NAME=your-bucket-name
GCS_CREDENTIALS_FILE=path/to/credentials.json
# Google Pub/Sub
PUBSUB_TOPIC=image-processing-topic
# Google Cloud Vision
VISION_API_ENABLED=true
# Security
API_KEY_SECRET=your-secret-key
# Vector database (optional)
# Vector database
VECTOR_DB_API_KEY=your-pinecone-api-key
VECTOR_DB_ENVIRONMENT=your-pinecone-environment
VECTOR_DB_INDEX_NAME=image-embeddings
@ -181,6 +234,180 @@ docker compose up --build
## Additional Information
## Design Decisions
### Database Selection: Firestore
- **Document-oriented model**: Ideal for hierarchical team/user/image data structures with flexible schemas
- **Real-time capabilities**: Enables live updates for collaborative features
- **Automatic scaling**: Handles variable workloads without manual intervention
- **ACID transactions**: Ensures data integrity for critical operations
- **Security rules**: Granular access control at the document level
- **Seamless GCP integration**: Works well with other Google Cloud services
### Storage Solution: Google Cloud Storage
- **Object storage optimized for binary data**: Perfect for image files of varying sizes
- **Content-delivery capabilities**: Fast global access to images
- **Lifecycle management**: Automated rules for moving less-accessed images to cheaper storage tiers
- **Fine-grained access control**: Secure pre-signed URLs for temporary access
- **Versioning support**: Maintains image history when needed
- **Cost-effective**: Pay only for what you use with no minimum fees
### Decoupled Embedding Generation
We deliberately decoupled the image embedding process from the upload flow for several reasons:
1. **Upload responsiveness**: Users experience fast upload times since compute-intensive embedding generation happens asynchronously
2. **System resilience**: Upload service remains available even if embedding service experiences issues
3. **Independent scaling**: Each component can scale based on its specific resource needs
4. **Cost optimization**: Cloud Functions only run when needed, avoiding idle compute costs
5. **Processing flexibility**: Can modify embedding algorithms without affecting the core upload flow
6. **Batch processing**: Potential to batch embedding generation for further cost optimization
### Latency Considerations
- **API response times**: FastAPI provides high-performance request handling
- **Caching strategy**: Frequently accessed images and search results are cached
- **Edge deployment**: Cloud Run regional deployment optimizes for user location
- **Async processing**: Non-blocking operations for concurrent request handling
- **Embedding pre-computation**: All embeddings are generated ahead of time, making searches fast
- **Search optimization**: Vector database indices are optimized for quick similarity searches
### Cost Optimization
- **Serverless architecture**: Pay-per-use model eliminates idle infrastructure costs
- **Storage tiering**: Automatic migration of older images to cheaper storage classes
- **Compute efficiency**: Cloud Functions minimize compute costs through precise scaling
- **Caching**: Reduces repeated processing of the same data
- **Resource throttling**: Rate limits prevent unexpected usage spikes
- **Embedding dimensions**: Balancing vector size for accuracy vs. storage costs
- **Query optimization**: Efficient search patterns to minimize vector database operations
### Scalability Approach
- **Horizontal scaling**: All components can scale out rather than up
- **Stateless design**: API servers maintain no local state, enabling easy replication
- **Queue-based workload distribution**: Prevents system overload during traffic spikes
- **Database sharding capability**: Firestore automatically shards data for growth
- **Vector database partitioning**: Pinecone handles distributed vector search at scale
- **Load balancing**: Traffic distributed across multiple service instances
- **Microservice architecture**: Individual components can scale independently based on demand
### Security Architecture
- **API key authentication**: Simple but effective access control for machine-to-machine communication
- **Team-based permissions**: Multi-tenant isolation with hierarchical access controls
- **Encrypted storage**: All data encrypted at rest and in transit
- **Secret management**: Sensitive configuration isolated from application code
- **Minimal attack surface**: Limited public endpoints with appropriate rate limiting
- **Audit logging**: Comprehensive activity tracking for security analysis
### Database Structure
The Firestore database is organized into collections and documents with the following structure:
#### Collections and Documents
```
firestore-root/
├── teams/ # Teams collection
│ └── {team_id}/ # Team document
│ ├── name: string # Team name
│ ├── created_at: timestamp
│ ├── updated_at: timestamp
│ │
│ ├── users/ # Team users subcollection
│ │ └── {user_id}/ # User document in team context
│ │ ├── role: string (admin, member, viewer)
│ │ ├── joined_at: timestamp
│ │ └── status: string (active, inactive)
│ │
│ ├── api_keys/ # Team API keys subcollection
│ │ └── {api_key_id}/ # API key document
│ │ ├── key_hash: string # Hashed API key value
│ │ ├── name: string # Key name/description
│ │ ├── created_at: timestamp
│ │ ├── expires_at: timestamp
│ │ ├── created_by: user_id
│ │ └── permissions: array # Specific permissions
│ │
│ └── collections/ # Image collections subcollection
│ └── {collection_id}/ # Collection document
│ ├── name: string
│ ├── description: string
│ ├── created_at: timestamp
│ ├── created_by: user_id
│ └── metadata: map # Collection-level metadata
│
├── users/ # Global users collection
│ └── {user_id}/ # User document
│ ├── email: string
│ ├── name: string
│ ├── created_at: timestamp
│ └── settings: map # User preferences
│
└── images/ # Images collection
└── {image_id}/ # Image document
├── filename: string
├── storage_path: string # GCS path
├── mime_type: string
├── size_bytes: number
├── width: number
├── height: number
├── uploaded_at: timestamp
├── uploaded_by: user_id
├── team_id: string
├── collection_id: string # Optional parent collection
├── status: string (processing, ready, error)
├── embedding_id: string # Reference to vector DB
├── metadata: map # Extracted and custom metadata
│ ├── labels: array # AI-generated labels
│ ├── colors: array # Dominant colors
│ ├── objects: array # Detected objects
│ ├── custom: map # User-defined metadata
│ └── exif: map # Original image EXIF data
│
└── processing/ # Processing subcollection
└── {job_id}/ # Processing job document
├── type: string # embedding, analysis, etc.
├── status: string # pending, running, complete, error
├── created_at: timestamp
├── updated_at: timestamp
├── completed_at: timestamp
└── error: string # Error message if applicable
```
#### Key Relationships and Indexes
- **Team-User**: Many-to-many relationship through the team's users subcollection
- **Team-Image**: One-to-many relationship (images belong to one team)
- **Collection-Image**: One-to-many relationship (images can belong to one collection)
- **User-Image**: One-to-many relationship (upload attribution)
#### Composite Indexes
The following composite indexes are created to support efficient queries:
1. `images` collection:
- `team_id` ASC, `uploaded_at` DESC → List recent images for a team
- `team_id` ASC, `collection_id` ASC, `uploaded_at` DESC → List recent images in a collection
- `team_id` ASC, `status` ASC, `uploaded_at` ASC → Find oldest processing images
- `uploaded_by` ASC, `uploaded_at` DESC → List user's recent uploads
2. `users` subcollection (within teams):
- `role` ASC, `joined_at` DESC → List team members by role
#### Security Rules
Firestore security rules enforce the following access patterns:
- Team admins can read/write all team data
- Team members can read all team data but can only write to collections and images
- Team viewers can only read team data
- Users can only access teams they belong to
- API keys have scoped access based on their assigned permissions
## License
This project is licensed under the MIT License - see the LICENSE file for details.

255
tests/api/test_image_upload.py Normal file
View File

@ -0,0 +1,255 @@
import os
import pytest
import uuid
from fastapi.testclient import TestClient
from unittest.mock import patch, MagicMock
from src.db.repositories.image_repository import ImageRepository, image_repository
from src.db.models.image import ImageModel
from main import app
# Hardcoded API key as requested
API_KEY = "Wwg4eJjJ.d03970d43cf3a454ad4168b3226b423f"
# Mock team ID for testing
MOCK_TEAM_ID = "test-team-123"
MOCK_USER_ID = "test-user-456"
@pytest.fixture
def test_image_path():
"""Get path to test image"""
# Assuming image.png exists in the images directory
return os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "images", "image.png")
@pytest.fixture
def client():
"""Create a test client"""
return TestClient(app)
@pytest.fixture
def mock_auth():
"""Mock the authentication to use our hardcoded API key"""
with patch('src.api.v1.auth.get_current_user') as mock_auth:
# Configure the mock to return a valid user
mock_auth.return_value = {
"id": MOCK_USER_ID,
"team_id": MOCK_TEAM_ID,
"email": "test@example.com",
"name": "Test User"
}
yield mock_auth
@pytest.fixture
def mock_storage_service():
"""Mock the storage service"""
with patch('src.services.storage.StorageService') as MockStorageService:
# Configure the mock
mock_service = MagicMock()
# Mock the upload_file method
test_storage_path = f"{MOCK_TEAM_ID}/test-image-{uuid.uuid4().hex}.png"
mock_service.upload_file.return_value = (
test_storage_path, # storage_path
"image/png", # content_type
1024, # file_size
{ # metadata
"width": 800,
"height": 600,
"format": "PNG",
"mode": "RGBA"
}
)
# Make the constructor return our mock
MockStorageService.return_value = mock_service
yield mock_service
@pytest.mark.asyncio
async def test_upload_image_endpoint(client, test_image_path, mock_auth, mock_storage_service):
"""Test the image upload endpoint"""
# First, implement a mock image repository for verification
with patch('src.db.repositories.image_repository.ImageRepository.create') as mock_create:
# Configure the mock to return a valid image model
mock_image = ImageModel(
filename="test-image.png",
original_filename="test_image.png",
file_size=1024,
content_type="image/png",
storage_path=f"{MOCK_TEAM_ID}/test-image-123.png",
team_id=MOCK_TEAM_ID,
uploader_id=MOCK_USER_ID
)
mock_create.return_value = mock_image
# Create API endpoint route if it doesn't exist yet
with patch('src.api.v1.images.router.post') as mock_post:
# Modify the router for testing purposes
async def mock_upload_image_handler(file, description=None, tags=None, current_user=None):
# This simulates the handler that would be in src/api/v1/images.py
# Store image in database
image = ImageModel(
filename="test-image.png",
original_filename="test_image.png",
file_size=1024,
content_type="image/png",
storage_path=f"{MOCK_TEAM_ID}/test-image-123.png",
team_id=MOCK_TEAM_ID,
uploader_id=MOCK_USER_ID,
description=description,
tags=tags.split(",") if tags else []
)
created_image = await image_repository.create(image)
# Return response
return {
"id": str(created_image.id),
"filename": created_image.filename,
"storage_path": created_image.storage_path,
"content_type": created_image.content_type,
"team_id": str(created_image.team_id),
"uploader_id": str(created_image.uploader_id),
"description": created_image.description,
"tags": created_image.tags
}
mock_post.return_value = mock_upload_image_handler
# Open the test image
with open(test_image_path, "rb") as f:
# Create the file upload
files = {"file": ("test_image.png", f, "image/png")}
# Make the request with our hardcoded API key
response = client.post(
"/api/v1/images",
headers={"X-API-Key": API_KEY},
files=files,
data={
"description": "Test image upload",
"tags": "test,upload,image"
}
)
# Verify the response
assert response.status_code == 200
# If API is not yet implemented, it will return our mock response
# If implemented, it should still verify these fields
data = response.json()
if isinstance(data, dict) and "id" in data:
assert "filename" in data
assert "storage_path" in data
assert "content_type" in data
assert data.get("team_id") == MOCK_TEAM_ID
assert data.get("uploader_id") == MOCK_USER_ID
# Verify the image was stored in the database
mock_create.assert_called_once()
@pytest.mark.asyncio
async def test_image_lifecycle(client, test_image_path, mock_auth, mock_storage_service):
"""Test the complete image lifecycle: upload, get, delete"""
# First, implement a mock image repository
with patch('src.db.repositories.image_repository.ImageRepository.create') as mock_create, \
patch('src.db.repositories.image_repository.ImageRepository.get_by_id') as mock_get, \
patch('src.db.repositories.image_repository.ImageRepository.delete') as mock_delete:
# Configure the mocks
test_image_id = "60f1e5b5e85d8b2b2c9b1c1f" # mock ObjectId
test_storage_path = f"{MOCK_TEAM_ID}/test-image-123.png"
mock_image = ImageModel(
id=test_image_id,
filename="test-image.png",
original_filename="test_image.png",
file_size=1024,
content_type="image/png",
storage_path=test_storage_path,
team_id=MOCK_TEAM_ID,
uploader_id=MOCK_USER_ID,
description="Test image upload",
tags=["test", "upload", "image"]
)
mock_create.return_value = mock_image
mock_get.return_value = mock_image
mock_delete.return_value = True
# Mock the image API endpoints for a complete lifecycle test
with patch('src.api.v1.images.router.post') as mock_post, \
patch('src.api.v1.images.router.get') as mock_get_api, \
patch('src.api.v1.images.router.delete') as mock_delete_api:
# Mock the endpoints
async def mock_upload_handler(file, description=None, tags=None, current_user=None):
created_image = await image_repository.create(mock_image)
return {
"id": str(created_image.id),
"filename": created_image.filename,
"storage_path": created_image.storage_path,
"content_type": created_image.content_type,
"team_id": str(created_image.team_id),
"uploader_id": str(created_image.uploader_id),
"description": created_image.description,
"tags": created_image.tags
}
async def mock_get_handler(image_id, current_user=None):
image = await image_repository.get_by_id(image_id)
return {
"id": str(image.id),
"filename": image.filename,
"storage_path": image.storage_path,
"content_type": image.content_type,
"team_id": str(image.team_id),
"uploader_id": str(image.uploader_id),
"description": image.description,
"tags": image.tags
}
async def mock_delete_handler(image_id, current_user=None):
success = await image_repository.delete(image_id)
return {"success": success}
mock_post.return_value = mock_upload_handler
mock_get_api.return_value = mock_get_handler
mock_delete_api.return_value = mock_delete_handler
# 1. UPLOAD IMAGE
with open(test_image_path, "rb") as f:
response_upload = client.post(
"/api/v1/images",
headers={"X-API-Key": API_KEY},
files={"file": ("test_image.png", f, "image/png")},
data={"description": "Test image upload", "tags": "test,upload,image"}
)
# Verify upload
assert response_upload.status_code == 200
upload_data = response_upload.json()
image_id = upload_data.get("id")
assert image_id
# 2. GET IMAGE
response_get = client.get(
f"/api/v1/images/{image_id}",
headers={"X-API-Key": API_KEY}
)
# Verify get
assert response_get.status_code == 200
get_data = response_get.json()
assert get_data["id"] == image_id
assert get_data["filename"] == "test-image.png"
# 3. DELETE IMAGE
response_delete = client.delete(
f"/api/v1/images/{image_id}",
headers={"X-API-Key": API_KEY}
)
# Verify delete
assert response_delete.status_code == 200
delete_data = response_delete.json()
assert delete_data["success"] is True

192
tests/services/test_storage_service.py Normal file
View File

@ -0,0 +1,192 @@
import os
import pytest
import uuid
from fastapi import UploadFile
from unittest.mock import patch, MagicMock
from io import BytesIO
from src.services.storage import StorageService
from src.db.repositories.image_repository import ImageRepository, image_repository
from src.db.models.image import ImageModel
# Hardcoded API key as requested
API_KEY = "Wwg4eJjJ.d03970d43cf3a454ad4168b3226b423f"
# Mock team ID for testing
MOCK_TEAM_ID = "test-team-123"
@pytest.fixture
def test_image_path():
"""Get path to test image"""
# Assuming image.png exists in the images directory
return os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "images", "image.png")
@pytest.fixture
def test_image_data(test_image_path):
"""Get test image data"""
with open(test_image_path, "rb") as f:
return f.read()
@pytest.fixture
def test_upload_file(test_image_data):
"""Create a test UploadFile object"""
file = UploadFile(
filename="test_image.png",
file=BytesIO(test_image_data),
content_type="image/png"
)
return file
@pytest.mark.asyncio
async def test_upload_image_and_verify():
"""Test uploading an image and verifying it was added to storage and database"""
# Create mocks
mock_storage_client = MagicMock()
mock_bucket = MagicMock()
mock_blob = MagicMock()
# Configure mocks
mock_storage_client.bucket.return_value = mock_bucket
mock_bucket.exists.return_value = True
mock_bucket.blob.return_value = mock_blob
mock_blob.exists.return_value = True
# Generate a unique filename for the test
test_filename = f"test-{uuid.uuid4().hex}.png"
test_content = b"test image content"
test_content_type = "image/png"
test_file_size = len(test_content)
# Create a test upload file
upload_file = UploadFile(
filename=test_filename,
file=BytesIO(test_content),
content_type=test_content_type
)
# Patch the storage client
with patch('src.services.storage.StorageService._create_storage_client', return_value=mock_storage_client), \
patch('src.services.storage.StorageService._get_or_create_bucket', return_value=mock_bucket), \
patch('src.db.repositories.image_repository.ImageRepository.create') as mock_create:
# Configure the mock to return a valid image model
storage_path = f"{MOCK_TEAM_ID}/{test_filename}"
mock_image = ImageModel(
filename=test_filename,
original_filename=test_filename,
file_size=test_file_size,
content_type=test_content_type,
storage_path=storage_path,
team_id=MOCK_TEAM_ID,
uploader_id="test-user-123"
)
mock_create.return_value = mock_image
# Create a storage service instance
storage_service = StorageService()
# Upload the file
storage_path, content_type, file_size, metadata = await storage_service.upload_file(
upload_file, MOCK_TEAM_ID
)
# Verify the file was uploaded
mock_bucket.blob.assert_called_with(storage_path)
mock_blob.upload_from_string.assert_called_once()
# Verify storage path and content type
assert storage_path.startswith(MOCK_TEAM_ID)
assert content_type == test_content_type
assert file_size == test_file_size
@pytest.mark.asyncio
async def test_upload_and_retrieve_image():
"""Test uploading an image and then retrieving it"""
# Create mocks
mock_storage_client = MagicMock()
mock_bucket = MagicMock()
mock_blob = MagicMock()
# Configure mocks
mock_storage_client.bucket.return_value = mock_bucket
mock_bucket.exists.return_value = True
mock_bucket.blob.return_value = mock_blob
mock_blob.exists.return_value = True
# Set up the blob to return test content when downloaded
test_content = b"test image content"
mock_blob.download_as_bytes.return_value = test_content
# Generate a unique filename for the test
test_filename = f"test-{uuid.uuid4().hex}.png"
test_content_type = "image/png"
test_file_size = len(test_content)
# Create a test upload file
upload_file = UploadFile(
filename=test_filename,
file=BytesIO(test_content),
content_type=test_content_type
)
# Patch the storage client
with patch('src.services.storage.StorageService._create_storage_client', return_value=mock_storage_client), \
patch('src.services.storage.StorageService._get_or_create_bucket', return_value=mock_bucket):
# Create a storage service instance
storage_service = StorageService()
# Upload the file
storage_path, content_type, file_size, metadata = await storage_service.upload_file(
upload_file, MOCK_TEAM_ID
)
# Retrieve the file
retrieved_content = storage_service.get_file(storage_path)
# Verify the content was retrieved
mock_blob.download_as_bytes.assert_called_once()
assert retrieved_content == test_content
@pytest.mark.asyncio
async def test_upload_with_real_image(test_upload_file):
"""Test uploading a real image file (requires proper configuration)"""
# This test requires actual credentials and configuration
# It will be skipped by default to avoid failures in CI environments
# Skip if no API key environment is configured
if not API_KEY:
pytest.skip("API key not configured")
# Create mocks for storage
mock_storage_client = MagicMock()
mock_bucket = MagicMock()
mock_blob = MagicMock()
# Configure mocks
mock_storage_client.bucket.return_value = mock_bucket
mock_bucket.exists.return_value = True
mock_bucket.blob.return_value = mock_blob
# Patch the storage client
with patch('src.services.storage.StorageService._create_storage_client', return_value=mock_storage_client), \
patch('src.services.storage.StorageService._get_or_create_bucket', return_value=mock_bucket), \
patch('src.db.repositories.image_repository.ImageRepository.create') as mock_create:
# Create a storage service instance
storage_service = StorageService()
# Upload the real image file
storage_path, content_type, file_size, metadata = await storage_service.upload_file(
test_upload_file, MOCK_TEAM_ID
)
# Verify the file was uploaded
assert storage_path.startswith(MOCK_TEAM_ID)
assert content_type == "image/png"
assert file_size > 0
# Check if metadata was extracted (if it's an image)
if content_type.startswith('image/'):
assert 'width' in metadata
assert 'height' in metadata