Files
TimeTracker/app/utils/file_upload.py
T
Dries Peeters 9d1ece5263 feat: Implement comprehensive architectural improvements and new features
This commit implements a complete architectural transformation of the TimeTracker
application, introducing modern design patterns and comprehensive feature set.

## Architecture Improvements

### Service Layer (18 Services)
- TimeTrackingService: Time entry management with timer functionality
- ProjectService: Project operations and lifecycle management
- InvoiceService: Invoice creation, management, and status tracking
- TaskService: Task management and workflow
- ExpenseService: Expense tracking and categorization
- ClientService: Client relationship management
- PaymentService: Payment processing and invoice reconciliation
- CommentService: Comment system for projects, tasks, and quotes
- UserService: User management and role operations
- NotificationService: Notification delivery system
- ReportingService: Report generation and analytics
- AnalyticsService: Event tracking and analytics
- ExportService: CSV export functionality
- ImportService: CSV import with validation
- EmailService: Email operations and invoice delivery
- PermissionService: Role-based permission management
- BackupService: Database backup operations
- HealthService: System health checks and monitoring

### Repository Layer (9 Repositories)
- BaseRepository: Generic CRUD operations
- TimeEntryRepository: Time entry data access
- ProjectRepository: Project data access with filtering
- InvoiceRepository: Invoice queries and status management
- TaskRepository: Task data access
- ExpenseRepository: Expense data access
- ClientRepository: Client data access
- UserRepository: User data access
- PaymentRepository: Payment data access
- CommentRepository: Comment data access

### Schema Layer (9 Schemas)
- Marshmallow schemas for validation and serialization
- Create, update, and full schemas for all entities
- Input validation and data transformation

### Utility Modules (15 Utilities)
- api_responses: Standardized API response helpers
- validation: Input validation utilities
- query_optimization: N+1 query prevention and eager loading
- error_handlers: Centralized error handling
- cache: Caching foundation (Redis-ready)
- transactions: Transaction management decorators
- event_bus: Domain event system
- performance: Performance monitoring decorators
- logger: Enhanced structured logging
- pagination: Pagination utilities
- file_upload: Secure file upload handling
- search: Full-text search utilities
- rate_limiting: Rate limiting helpers
- config_manager: Configuration management
- datetime_utils: Enhanced date/time utilities

## Database Improvements
- Performance indexes migration (15+ indexes)
- Query optimization utilities
- N+1 query prevention patterns

## Testing Infrastructure
- Comprehensive test fixtures (conftest.py)
- Service layer unit tests
- Repository layer unit tests
- Integration test examples

## CI/CD Pipeline
- GitHub Actions workflow
- Automated linting (Black, Flake8, Pylint)
- Security scanning (Bandit, Safety, Semgrep)
- Automated testing with coverage
- Docker image builds

## Documentation
- Architecture migration guide
- Quick start guide
- API enhancements documentation
- Implementation summaries
- Refactored route examples

## Key Benefits
- Separation of concerns: Business logic decoupled from routes
- Testability: Services and repositories can be tested in isolation
- Maintainability: Consistent patterns across codebase
- Performance: Database indexes and query optimization
- Security: Input validation and security scanning
- Scalability: Event-driven architecture and health checks

## Statistics
- 70+ new files created
- 8,000+ lines of code
- 18 services, 9 repositories, 9 schemas
- 15 utility modules
- 5 test files with examples

This transformation establishes a solid foundation for future development
and follows industry best practices for maintainable, scalable applications.
2025-11-23 20:00:10 +01:00

161 lines
4.3 KiB
Python

"""
File upload utilities with validation and security.
"""
from typing import Optional, Tuple
from werkzeug.utils import secure_filename
from flask import current_app
import os
from pathlib import Path
from app.constants import (
MAX_FILE_SIZE,
ALLOWED_IMAGE_EXTENSIONS,
ALLOWED_DOCUMENT_EXTENSIONS
)
def validate_file_upload(
file,
allowed_extensions: Optional[set] = None,
max_size: int = MAX_FILE_SIZE
) -> Tuple[bool, Optional[str]]:
"""
Validate a file upload.
Args:
file: File object from request
allowed_extensions: Set of allowed extensions (defaults to all)
max_size: Maximum file size in bytes
Returns:
tuple of (is_valid, error_message)
"""
if not file or not file.filename:
return False, "No file provided"
# Check file size
file.seek(0, os.SEEK_END)
file_size = file.tell()
file.seek(0)
if file_size > max_size:
return False, f"File size exceeds maximum of {max_size / (1024*1024):.1f}MB"
# Check extension
if allowed_extensions:
filename = secure_filename(file.filename)
ext = Path(filename).suffix.lower()
if ext not in allowed_extensions:
return False, f"File type not allowed. Allowed types: {', '.join(allowed_extensions)}"
return True, None
def save_uploaded_file(
file,
upload_folder: str,
subfolder: Optional[str] = None,
prefix: Optional[str] = None
) -> Optional[str]:
"""
Save an uploaded file securely.
Args:
file: File object from request
upload_folder: Base upload folder
subfolder: Optional subfolder (e.g., 'receipts', 'avatars')
prefix: Optional filename prefix
Returns:
Saved file path or None on error
"""
try:
# Secure filename
filename = secure_filename(file.filename)
if not filename:
return None
# Add prefix if provided
if prefix:
name, ext = os.path.splitext(filename)
filename = f"{prefix}_{name}{ext}"
# Create directory structure
if subfolder:
upload_path = os.path.join(upload_folder, subfolder)
else:
upload_path = upload_folder
os.makedirs(upload_path, exist_ok=True)
# Ensure unique filename
filepath = os.path.join(upload_path, filename)
counter = 1
while os.path.exists(filepath):
name, ext = os.path.splitext(filename)
filepath = os.path.join(upload_path, f"{name}_{counter}{ext}")
counter += 1
# Save file
file.save(filepath)
# Return relative path
if subfolder:
return os.path.join(subfolder, os.path.basename(filepath))
return os.path.basename(filepath)
except Exception as e:
current_app.logger.error(f"Error saving uploaded file: {e}")
return None
def delete_uploaded_file(filepath: str, upload_folder: str) -> bool:
"""
Delete an uploaded file.
Args:
filepath: Relative file path
upload_folder: Base upload folder
Returns:
True if deleted, False otherwise
"""
try:
full_path = os.path.join(upload_folder, filepath)
if os.path.exists(full_path):
os.remove(full_path)
return True
return False
except Exception as e:
current_app.logger.error(f"Error deleting file {filepath}: {e}")
return False
def get_file_info(filepath: str, upload_folder: str) -> Optional[dict]:
"""
Get information about an uploaded file.
Args:
filepath: Relative file path
upload_folder: Base upload folder
Returns:
dict with file info or None
"""
try:
full_path = os.path.join(upload_folder, filepath)
if not os.path.exists(full_path):
return None
stat = os.stat(full_path)
return {
'path': filepath,
'size': stat.st_size,
'modified': stat.st_mtime,
'extension': Path(filepath).suffix.lower()
}
except Exception as e:
current_app.logger.error(f"Error getting file info: {e}")
return None