File size: 3,337 Bytes
a10e62e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 | import logging
from typing import Any, Dict, List, Optional
from fastapi import Depends, Request
from pydantic import BaseModel
from sqlalchemy.orm import Session
from core.api_governance import ActionComplexity, require_governance
from core.auth import get_current_user
from core.base_routes import BaseAPIRouter
from core.connection_service import connection_service
from core.database import get_db
from core.models import User
router = BaseAPIRouter(prefix="/api/v1/connections", tags=["Connections"])
logger = logging.getLogger(__name__)
class ConnectionResponse(BaseModel):
id: str
name: str
integration_id: str
status: str
created_at: Optional[str] = None
last_used: Optional[str] = None
@router.get("/", response_model=List[ConnectionResponse])
async def list_connections(integration_id: Optional[str] = None, current_user: User = Depends(get_current_user)):
return connection_service.get_connections(current_user.id, integration_id)
@router.delete("/{connection_id}")
@require_governance(
action_complexity=ActionComplexity.HIGH,
action_name="delete_connection",
feature="connection"
)
async def delete_connection(
connection_id: str,
request: Request,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db),
agent_id: Optional[str] = None
):
"""
Delete a connection.
**Governance**: Requires SUPERVISED+ maturity (HIGH complexity).
- Connection deletion is a state-changing operation
- Requires SUPERVISED maturity or higher
"""
success = connection_service.delete_connection(connection_id, current_user.id)
if not success:
raise router.not_found_error("Connection", connection_id)
logger.info(f"Connection deleted: {connection_id}")
return router.success_response(message="Connection deleted successfully")
class RenameConnectionRequest(BaseModel):
name: str
@router.patch("/{connection_id}")
@require_governance(
action_complexity=ActionComplexity.MODERATE,
action_name="rename_connection",
feature="connection"
)
async def rename_connection(
connection_id: str,
req: RenameConnectionRequest,
request: Request,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db),
agent_id: Optional[str] = None
):
"""
Rename a connection.
**Governance**: Requires INTERN+ maturity (MODERATE complexity).
- Connection modification is a moderate action
- Requires INTERN maturity or higher
"""
success = connection_service.update_connection_name(connection_id, current_user.id, req.name)
if not success:
raise router.not_found_error("Connection", connection_id)
logger.info(f"Connection renamed: {connection_id}")
return router.success_response(message="Connection renamed successfully")
@router.get("/{connection_id}/credentials")
async def get_credentials(connection_id: str, current_user: User = Depends(get_current_user)):
"""
Internal use only / Dev only. In production, we should never expose raw credentials.
"""
creds = connection_service.get_connection_credentials(connection_id, current_user.id)
if not creds:
raise router.not_found_error("Connection", connection_id)
return router.success_response(data=creds, message="Credentials retrieved")
|