79 lines
2.1 KiB
Python
79 lines
2.1 KiB
Python
from http import HTTPStatus
|
|
from typing import Annotated
|
|
|
|
from aws_lambda_powertools.event_handler.api_gateway import Router
|
|
from aws_lambda_powertools.event_handler.openapi.params import Body
|
|
from layercake.dateutils import now
|
|
from layercake.dynamodb import DynamoDBPersistenceLayer, KeyPair
|
|
from layercake.extra_types import NameStr
|
|
from pydantic import BaseModel, EmailStr
|
|
|
|
from api_gateway import JSONResponse
|
|
from boto3clients import dynamodb_client
|
|
from config import USER_TABLE
|
|
|
|
router = Router()
|
|
user_layer = DynamoDBPersistenceLayer(USER_TABLE, dynamodb_client)
|
|
|
|
|
|
@router.get('/<id>/admins', compress=True)
|
|
def admins(id: str):
|
|
return user_layer.collection.query(
|
|
KeyPair(id, 'admins'),
|
|
limit=100,
|
|
)
|
|
|
|
|
|
class Admin(BaseModel):
|
|
id: str
|
|
name: NameStr
|
|
email: EmailStr
|
|
|
|
|
|
@router.post('/<id>/admins', compress=True)
|
|
def grant(id: str, admin: Admin):
|
|
now_ = now()
|
|
|
|
with user_layer.transact_writer() as transact:
|
|
transact.condition(
|
|
key=KeyPair(f'orgmembers#{id}', admin.id),
|
|
cond_expr='attribute_exists(sk)',
|
|
)
|
|
# Grant admin privileges
|
|
transact.put(
|
|
item={
|
|
'id': admin.id,
|
|
'sk': f'acls#{id}',
|
|
'roles': ['ADMIN'],
|
|
'created_ad': now_,
|
|
},
|
|
# cond_expr='attribute_not_exists(sk)',
|
|
)
|
|
# Add user to admin list
|
|
transact.put(
|
|
item={
|
|
'id': id,
|
|
'sk': f'admins#{admin.id}',
|
|
'name': admin.name,
|
|
'email': admin.email,
|
|
'created_ad': now_,
|
|
},
|
|
cond_expr='attribute_not_exists(sk)',
|
|
)
|
|
|
|
return JSONResponse(status_code=HTTPStatus.CREATED)
|
|
|
|
|
|
@router.delete('/<id>/admins', compress=True)
|
|
def revoke(
|
|
id: str,
|
|
user_id: Annotated[str, Body(embed=True)],
|
|
):
|
|
with user_layer.transact_writer() as transact:
|
|
# Revoke admin privileges
|
|
transact.delete(key=KeyPair(user_id, f'acls#{id}'))
|
|
# Remove user from admin list
|
|
transact.delete(key=KeyPair(id, f'admins#{user_id}'))
|
|
|
|
return JSONResponse(status_code=HTTPStatus.OK)
|