nikdoof/leggen
1
1
Fork 0
mirror of https://github.com/elisiariocouto/leggen.git synced 2025-12-23 20:49:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

Lint and reformat.

Browse Source
This commit is contained in:
Elisiário Couto
2025-09-28 23:01:04 +01:00
committed by Elisiário Couto
parent 22ec0e36b1
commit 222bb2ec64
7 changed files with 242 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
leggen/api/models/backup.py
View File

@@ -7,32 +7,34 @@ from pydantic import BaseModel, Field
class S3Config(BaseModel):
"""S3 backup configuration model for API."""
access_key_id: str = Field(..., description="AWS S3 access key ID")
secret_access_key: str = Field(..., description="AWS S3 secret access key")
bucket_name: str = Field(..., description="S3 bucket name")
region: str = Field(default="us-east-1", description="AWS S3 region")
endpoint_url: Optional[str] = Field(default=None, description="Custom S3 endpoint URL")
endpoint_url: Optional[str] = Field(
default=None, description="Custom S3 endpoint URL"
)
path_style: bool = Field(default=False, description="Use path-style addressing")
enabled: bool = Field(default=True, description="Enable S3 backups")
class BackupSettings(BaseModel):
"""Backup settings model for API."""
s3: Optional[S3Config] = None
class BackupTest(BaseModel):
"""Backup connection test request model."""
service: str = Field(..., description="Backup service type (s3)")
config: S3Config = Field(..., description="S3 configuration to test")
class BackupInfo(BaseModel):
"""Backup file information model."""
key: str = Field(..., description="S3 object key")
last_modified: str = Field(..., description="Last modified timestamp (ISO format)")
size: int = Field(..., description="File size in bytes")
@@ -40,6 +42,8 @@ class BackupInfo(BaseModel):
class BackupOperation(BaseModel):
"""Backup operation request model."""
operation: str = Field(..., description="Operation type (backup, restore)")
backup_key: Optional[str] = Field(default=None, description="Backup key for restore operations")
backup_key: Optional[str] = Field(
default=None, description="Backup key for restore operations"
)

70
leggen/api/routes/backup.py
View File

@@ -1,6 +1,5 @@
"""API routes for backup management."""
from fastapi import APIRouter, HTTPException
from loguru import logger
@@ -24,10 +23,10 @@ async def get_backup_settings() -> APIResponse:
"""Get current backup settings."""
try:
backup_config = config.backup_config
# Build response safely without exposing secrets
s3_config = backup_config.get("s3", {})
settings = BackupSettings(
s3=S3Config(
access_key_id="***" if s3_config.get("access_key_id") else "",
@@ -41,13 +40,13 @@ async def get_backup_settings() -> APIResponse:
if s3_config.get("bucket_name")
else None,
)
return APIResponse(
success=True,
data=settings,
message="Backup settings retrieved successfully",
)
except Exception as e:
logger.error(f"Failed to get backup settings: {e}")
raise HTTPException(
@@ -71,20 +70,20 @@ async def update_backup_settings(settings: BackupSettings) -> APIResponse:
path_style=settings.s3.path_style,
enabled=settings.s3.enabled,
)
# Test connection
backup_service = BackupService()
connection_success = await backup_service.test_connection(s3_config)
if not connection_success:
raise HTTPException(
status_code=400,
detail="S3 connection test failed. Please check your configuration."
detail="S3 connection test failed. Please check your configuration.",
)
# Update backup config
backup_config = {}
if settings.s3:
backup_config["s3"] = {
"access_key_id": settings.s3.access_key_id,
@@ -95,17 +94,17 @@ async def update_backup_settings(settings: BackupSettings) -> APIResponse:
"path_style": settings.s3.path_style,
"enabled": settings.s3.enabled,
}
# Save to config
if backup_config:
config.update_section("backup", backup_config)
return APIResponse(
success=True,
data={"updated": True},
message="Backup settings updated successfully",
)
except HTTPException:
raise
except Exception as e:
@@ -123,7 +122,7 @@ async def test_backup_connection(test_request: BackupTest) -> APIResponse:
raise HTTPException(
status_code=400, detail="Only 's3' service is supported"
)
# Convert API model to config model
s3_config = S3BackupConfig(
access_key_id=test_request.config.access_key_id,
@@ -134,10 +133,10 @@ async def test_backup_connection(test_request: BackupTest) -> APIResponse:
path_style=test_request.config.path_style,
enabled=test_request.config.enabled,
)
backup_service = BackupService()
success = await backup_service.test_connection(s3_config)
if success:
return APIResponse(
success=True,
@@ -149,7 +148,7 @@ async def test_backup_connection(test_request: BackupTest) -> APIResponse:
success=False,
message="S3 connection test failed",
)
except HTTPException:
raise
except Exception as e:
@@ -164,26 +163,26 @@ async def list_backups() -> APIResponse:
"""List available backups."""
try:
backup_config = config.backup_config.get("s3", {})
if not backup_config.get("bucket_name"):
return APIResponse(
success=True,
data=[],
message="No S3 backup configuration found",
)
# Convert config to model
s3_config = S3BackupConfig(**backup_config)
backup_service = BackupService(s3_config)
backups = await backup_service.list_backups()
return APIResponse(
success=True,
data=backups,
message=f"Found {len(backups)} backups",
)
except Exception as e:
logger.error(f"Failed to list backups: {e}")
raise HTTPException(
@@ -196,12 +195,10 @@ async def backup_operation(operation_request: BackupOperation) -> APIResponse:
"""Perform backup operation (backup or restore)."""
try:
backup_config = config.backup_config.get("s3", {})
if not backup_config.get("bucket_name"):
raise HTTPException(
status_code=400, detail="S3 backup is not configured"
)
raise HTTPException(status_code=400, detail="S3 backup is not configured")
# Convert config to model with validation
try:
s3_config = S3BackupConfig(**backup_config)
@@ -209,14 +206,14 @@ async def backup_operation(operation_request: BackupOperation) -> APIResponse:
raise HTTPException(
status_code=400, detail=f"Invalid S3 configuration: {str(e)}"
) from e
backup_service = BackupService(s3_config)
if operation_request.operation == "backup":
# Backup database
database_path = path_manager.get_database_path()
success = await backup_service.backup_database(database_path)
if success:
return APIResponse(
success=True,
@@ -228,19 +225,20 @@ async def backup_operation(operation_request: BackupOperation) -> APIResponse:
success=False,
message="Database backup failed",
)
elif operation_request.operation == "restore":
if not operation_request.backup_key:
raise HTTPException(
status_code=400, detail="backup_key is required for restore operation"
status_code=400,
detail="backup_key is required for restore operation",
)
# Restore database
database_path = path_manager.get_database_path()
success = await backup_service.restore_database(
operation_request.backup_key, database_path
)
if success:
return APIResponse(
success=True,
@@ -256,11 +254,11 @@ async def backup_operation(operation_request: BackupOperation) -> APIResponse:
raise HTTPException(
status_code=400, detail="Invalid operation. Use 'backup' or 'restore'"
)
except HTTPException:
raise
except Exception as e:
logger.error(f"Failed to perform backup operation: {e}")
raise HTTPException(
status_code=500, detail=f"Failed to perform backup operation: {str(e)}"
) from e
) from e