mirror of
https://github.com/mealie-recipes/mealie.git
synced 2025-12-28 21:15:26 -05:00
23c039b42d
* refactored event dispatching added EventDocumentType and EventOperation to Event added event listeners to bulk recipe changes overhauled shopping list item events to be more useful modified shopping list item repo to return more information * added internal documentation for event types * renamed message_types.py to event_types.py * added unique event id and fixed instantiation * generalized event listeners and publishers moved apprise publisher to new apprise event listener fixed duplicate message bug with apprise publisher * added JWT field for user-specified integration id * removed obselete test notification route * tuned up existing notification tests * added dependency to get integration_id from jwt * added base crud controller to facilitate events * simplified event publishing * temporarily fixed test notification
56 lines
1.9 KiB
Python
56 lines
1.9 KiB
Python
from datetime import timedelta
|
|
|
|
from fastapi import HTTPException, status
|
|
|
|
from mealie.core.security import create_access_token
|
|
from mealie.routes._base import BaseUserController, controller
|
|
from mealie.routes._base.routers import UserAPIRouter
|
|
from mealie.schema.user import CreateToken, DeleteTokenResponse, LongLiveTokenIn, LongLiveTokenInDB, LongLiveTokenOut
|
|
|
|
router = UserAPIRouter(prefix="/users", tags=["Users: Tokens"])
|
|
|
|
|
|
@controller(router)
|
|
class UserApiTokensController(BaseUserController):
|
|
@router.post("/api-tokens", status_code=status.HTTP_201_CREATED, response_model=LongLiveTokenOut)
|
|
def create_api_token(
|
|
self,
|
|
token_params: LongLiveTokenIn,
|
|
):
|
|
"""Create api_token in the Database"""
|
|
|
|
token_data = {
|
|
"long_token": True,
|
|
"id": str(self.user.id),
|
|
"name": token_params.name,
|
|
"integration_id": token_params.integration_id,
|
|
}
|
|
|
|
five_years = timedelta(1825)
|
|
token = create_access_token(token_data, five_years)
|
|
|
|
token_model = CreateToken(
|
|
name=token_params.name,
|
|
token=token,
|
|
user_id=self.user.id,
|
|
)
|
|
|
|
new_token_in_db = self.repos.api_tokens.create(token_model)
|
|
|
|
if new_token_in_db:
|
|
return new_token_in_db
|
|
|
|
@router.delete("/api-tokens/{token_id}", response_model=DeleteTokenResponse)
|
|
def delete_api_token(self, token_id: int):
|
|
"""Delete api_token from the Database"""
|
|
token: LongLiveTokenInDB = self.repos.api_tokens.get_one(token_id)
|
|
|
|
if not token:
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, f"Could not locate token with id '{token_id}' in database")
|
|
|
|
if token.user.email == self.user.email:
|
|
deleted_token = self.repos.api_tokens.delete(token_id)
|
|
return DeleteTokenResponse(token_delete=deleted_token.name)
|
|
else:
|
|
raise HTTPException(status.HTTP_403_FORBIDDEN)
|