135 lines
4.1 KiB
Python
135 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
FastAPI application for DICOM migration service.
|
|
"""
|
|
import sys
|
|
import os
|
|
import json
|
|
from typing import Dict, Any, Optional
|
|
from fastapi import FastAPI, HTTPException, Depends, status
|
|
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
|
from pydantic import BaseModel
|
|
|
|
# Add parent directory to sys.path to allow importing from the main project
|
|
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
# Import the process module for API-specific processing
|
|
from process import process_study_by_accession
|
|
from utils.logger import setup_logger
|
|
from utils.dicom_utils import create_directory_if_not_exists
|
|
from api_cleanup import register_api_cleanup
|
|
|
|
# Create API logger
|
|
api_logger = setup_logger("api", "logs/api.log")
|
|
|
|
# Initialize FastAPI app
|
|
app = FastAPI(
|
|
title="DICOM Migration API",
|
|
description="API for migrating DICOM studies by accession number",
|
|
version="1.0.0"
|
|
)
|
|
|
|
# Security scheme for Bearer token authentication
|
|
security = HTTPBearer()
|
|
|
|
# Environment variable for API key (use a secure method in production)
|
|
API_TOKEN = os.environ.get("API_TOKEN", "$2y$05$o1Sfikmwynq76PmuBkJeROpS2WcD.Sh4lgrjohMicnlhBlGAt8UAq")
|
|
|
|
class MigrationResponse(BaseModel):
|
|
"""Response model for migration endpoints."""
|
|
success: bool
|
|
message: str
|
|
details: Optional[Dict[str, Any]] = None
|
|
|
|
def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
|
|
"""
|
|
Verify the authentication token.
|
|
|
|
Args:
|
|
credentials: Bearer token credentials
|
|
|
|
Returns:
|
|
bool: True if token is valid
|
|
|
|
Raises:
|
|
HTTPException: If token is invalid
|
|
"""
|
|
if credentials.scheme != "Bearer":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid authentication scheme. Bearer token required."
|
|
)
|
|
|
|
if credentials.credentials != API_TOKEN:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid token or expired token"
|
|
)
|
|
|
|
return True
|
|
|
|
@app.on_event("startup")
|
|
async def startup_event():
|
|
"""Initialize resources when the API starts."""
|
|
# Create necessary directories
|
|
create_directory_if_not_exists("logs")
|
|
|
|
# Use our custom API cleanup registration that avoids signal handlers
|
|
register_api_cleanup()
|
|
|
|
api_logger.info("API started successfully")
|
|
|
|
@app.on_event("shutdown")
|
|
async def shutdown_event():
|
|
"""Clean up resources when the API shuts down."""
|
|
api_logger.info("API shutting down, cleaning up resources...")
|
|
# Perform cleanup operations here
|
|
from utils.cleanup import cleanup_dicom_files
|
|
cleanup_dicom_files()
|
|
api_logger.info("Cleanup complete, API shutdown successful")
|
|
|
|
@app.get("/", response_model=MigrationResponse)
|
|
async def root():
|
|
"""
|
|
Root endpoint to check if the API is running.
|
|
"""
|
|
return {
|
|
"success": True,
|
|
"message": "DICOM Migration API is running",
|
|
"details": {
|
|
"version": "1.0.0",
|
|
"copyright": "PT. Sadhana Abiyasa Sampoerna",
|
|
"developer": "Mario and FX Padmanto"
|
|
}
|
|
}
|
|
|
|
@app.post("/migrate/{accession_number}", response_model=MigrationResponse)
|
|
async def migrate_study(
|
|
accession_number: str,
|
|
_: bool = Depends(verify_token)
|
|
):
|
|
"""
|
|
Migrate a single study by its accession number.
|
|
|
|
Args:
|
|
accession_number: The accession number of the study to migrate
|
|
|
|
Returns:
|
|
MigrationResponse: The result of the migration operation
|
|
"""
|
|
api_logger.info(f"Received migration request for accession number: {accession_number}")
|
|
|
|
# Process the study using the specialized API processing module
|
|
result = process_study_by_accession(accession_number)
|
|
|
|
if result['success']:
|
|
api_logger.info(f"Successfully processed migration for accession number: {accession_number}")
|
|
else:
|
|
api_logger.error(f"Failed to process migration for accession number: {accession_number}: {result['message']}")
|
|
|
|
return result
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)
|