Add access permission management API
Backend:
- Add AccessPermission schemas (Create, Update, Response)
- Add ProjectWithAccessResponse schema
- Add permission endpoints:
- GET /project/{name}/permissions - list permissions (admin only)
- POST /project/{name}/permissions - grant access (admin only)
- PUT /project/{name}/permissions/{username} - update access
- DELETE /project/{name}/permissions/{username} - revoke access
- GET /project/{name}/my-access - get current user's access level
Frontend:
- Add AccessLevel, AccessPermission types
- Add API functions for access management:
- getMyProjectAccess()
- listProjectPermissions()
- grantProjectAccess()
- updateProjectAccess()
- revokeProjectAccess()
This commit is contained in:
@@ -107,6 +107,9 @@ from .schemas import (
|
||||
APIKeyCreate,
|
||||
APIKeyResponse,
|
||||
APIKeyCreateResponse,
|
||||
AccessPermissionCreate,
|
||||
AccessPermissionUpdate,
|
||||
AccessPermissionResponse,
|
||||
)
|
||||
from .metadata import extract_metadata
|
||||
from .config import get_settings
|
||||
@@ -1190,6 +1193,159 @@ def delete_project(
|
||||
return None
|
||||
|
||||
|
||||
# Access Permission routes
|
||||
@router.get(
|
||||
"/api/v1/project/{project_name}/permissions",
|
||||
response_model=List[AccessPermissionResponse],
|
||||
)
|
||||
def list_project_permissions(
|
||||
project_name: str,
|
||||
db: Session = Depends(get_db),
|
||||
current_user: User = Depends(get_current_user),
|
||||
):
|
||||
"""
|
||||
List all access permissions for a project.
|
||||
Requires admin access to the project.
|
||||
"""
|
||||
project = check_project_access(db, project_name, current_user, "admin")
|
||||
|
||||
auth_service = AuthorizationService(db)
|
||||
permissions = auth_service.list_project_permissions(str(project.id))
|
||||
|
||||
return permissions
|
||||
|
||||
|
||||
@router.post(
|
||||
"/api/v1/project/{project_name}/permissions",
|
||||
response_model=AccessPermissionResponse,
|
||||
)
|
||||
def grant_project_access(
|
||||
project_name: str,
|
||||
permission: AccessPermissionCreate,
|
||||
db: Session = Depends(get_db),
|
||||
current_user: User = Depends(get_current_user),
|
||||
):
|
||||
"""
|
||||
Grant access to a user for a project.
|
||||
Requires admin access to the project.
|
||||
"""
|
||||
project = check_project_access(db, project_name, current_user, "admin")
|
||||
|
||||
auth_service = AuthorizationService(db)
|
||||
new_permission = auth_service.grant_access(
|
||||
str(project.id),
|
||||
permission.username,
|
||||
permission.level,
|
||||
permission.expires_at,
|
||||
)
|
||||
|
||||
return new_permission
|
||||
|
||||
|
||||
@router.put(
|
||||
"/api/v1/project/{project_name}/permissions/{username}",
|
||||
response_model=AccessPermissionResponse,
|
||||
)
|
||||
def update_project_access(
|
||||
project_name: str,
|
||||
username: str,
|
||||
permission: AccessPermissionUpdate,
|
||||
db: Session = Depends(get_db),
|
||||
current_user: User = Depends(get_current_user),
|
||||
):
|
||||
"""
|
||||
Update a user's access level for a project.
|
||||
Requires admin access to the project.
|
||||
"""
|
||||
project = check_project_access(db, project_name, current_user, "admin")
|
||||
|
||||
auth_service = AuthorizationService(db)
|
||||
|
||||
# Get existing permission
|
||||
from .models import AccessPermission
|
||||
existing = (
|
||||
db.query(AccessPermission)
|
||||
.filter(
|
||||
AccessPermission.project_id == project.id,
|
||||
AccessPermission.user_id == username,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
if not existing:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail=f"No access permission found for user '{username}'",
|
||||
)
|
||||
|
||||
# Update fields
|
||||
if permission.level is not None:
|
||||
existing.level = permission.level
|
||||
if permission.expires_at is not None:
|
||||
existing.expires_at = permission.expires_at
|
||||
|
||||
db.commit()
|
||||
db.refresh(existing)
|
||||
|
||||
return existing
|
||||
|
||||
|
||||
@router.delete(
|
||||
"/api/v1/project/{project_name}/permissions/{username}",
|
||||
status_code=204,
|
||||
)
|
||||
def revoke_project_access(
|
||||
project_name: str,
|
||||
username: str,
|
||||
db: Session = Depends(get_db),
|
||||
current_user: User = Depends(get_current_user),
|
||||
):
|
||||
"""
|
||||
Revoke a user's access to a project.
|
||||
Requires admin access to the project.
|
||||
"""
|
||||
project = check_project_access(db, project_name, current_user, "admin")
|
||||
|
||||
auth_service = AuthorizationService(db)
|
||||
deleted = auth_service.revoke_access(str(project.id), username)
|
||||
|
||||
if not deleted:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail=f"No access permission found for user '{username}'",
|
||||
)
|
||||
|
||||
return None
|
||||
|
||||
|
||||
@router.get(
|
||||
"/api/v1/project/{project_name}/my-access",
|
||||
)
|
||||
def get_my_project_access(
|
||||
project_name: str,
|
||||
db: Session = Depends(get_db),
|
||||
current_user: Optional[User] = Depends(get_current_user_optional),
|
||||
):
|
||||
"""
|
||||
Get the current user's access level for a project.
|
||||
Returns null for anonymous users on private projects.
|
||||
"""
|
||||
from .models import Project
|
||||
|
||||
project = db.query(Project).filter(Project.name == project_name).first()
|
||||
if not project:
|
||||
raise HTTPException(status_code=404, detail="Project not found")
|
||||
|
||||
auth_service = AuthorizationService(db)
|
||||
access_level = auth_service.get_user_access_level(str(project.id), current_user)
|
||||
|
||||
return {
|
||||
"project": project_name,
|
||||
"access_level": access_level,
|
||||
"is_owner": current_user and project.created_by == current_user.username,
|
||||
}
|
||||
|
||||
|
||||
# Package routes
|
||||
@router.get(
|
||||
"/api/v1/project/{project_name}/packages",
|
||||
|
||||
Reference in New Issue
Block a user