Coverage for /usr/local/lib/python3.12/site-packages/prefect/server/api/csrf_token.py: 72%

21 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 10:48 +0000

1from typing import TYPE_CHECKING 1a

2 

3from fastapi import Depends, Query 1a

4from starlette.exceptions import HTTPException 1a

5 

6from prefect._internal.compatibility.starlette import status 1a

7from prefect.logging import get_logger 1a

8from prefect.server import models, schemas 1a

9from prefect.server.database import PrefectDBInterface, provide_database_interface 1a

10from prefect.server.utilities.server import PrefectRouter 1a

11from prefect.settings import PREFECT_SERVER_CSRF_PROTECTION_ENABLED 1a

12 

13if TYPE_CHECKING: 13 ↛ 14line 13 didn't jump to line 14 because the condition on line 13 was never true1a

14 import logging 

15 

16logger: "logging.Logger" = get_logger("server.api") 1a

17 

18router: PrefectRouter = PrefectRouter(prefix="/csrf-token") 1a

19 

20 

21@router.get("") 1a

22async def create_csrf_token( 1a

23 db: PrefectDBInterface = Depends(provide_database_interface), 

24 client: str = Query(..., description="The client to create a CSRF token for"), 

25) -> schemas.core.CsrfToken: 

26 """Create or update a CSRF token for a client""" 

27 if PREFECT_SERVER_CSRF_PROTECTION_ENABLED.value() is False: 27 ↛ 33line 27 didn't jump to line 33 because the condition on line 27 was always true1bc

28 raise HTTPException( 1bc

29 status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, 

30 detail="CSRF protection is disabled.", 

31 ) 

32 

33 async with db.session_context(begin_transaction=True) as session: 

34 token = await models.csrf_token.create_or_update_csrf_token( 

35 session=session, client=client 

36 ) 

37 await models.csrf_token.delete_expired_tokens(session=session) 

38 

39 return token