prefect.server.models.csrf_token

create_or_update_csrf_token async

Create or update a CSRF token for a client. If the client already has a token, it will be updated.

Parameters:

Name	Type	Description	Default
session	AsyncSession	The database session	required
client	str	The client identifier	required

Returns:

Type	Description
CsrfToken	core.CsrfToken: The CSRF token

Source code in prefect/server/models/csrf_token.py

@db_injector
async def create_or_update_csrf_token(
    db: PrefectDBInterface,
    session: AsyncSession,
    client: str,
) -> core.CsrfToken:
    """Create or update a CSRF token for a client. If the client already has a
    token, it will be updated.

    Args:
        session (AsyncSession): The database session
        client (str): The client identifier

    Returns:
        core.CsrfToken: The CSRF token
    """

    expiration = (
        datetime.now(timezone.utc)
        + settings.PREFECT_SERVER_CSRF_TOKEN_EXPIRATION.value()
    )
    token = secrets.token_hex(32)

    await session.execute(
        db.insert(db.CsrfToken)
        .values(
            client=client,
            token=token,
            expiration=expiration,
        )
        .on_conflict_do_update(
            index_elements=[db.CsrfToken.client],
            set_={"token": token, "expiration": expiration},
        ),
    )

    # Return the created / updated token object
    return await read_token_for_client(session=session, client=client)

delete_expired_tokens async

Delete expired CSRF tokens.

Parameters:

Name	Type	Description	Default
session	AsyncSession	The database session	required

Returns:

Name	Type	Description
int	int	The number of tokens deleted

Source code in prefect/server/models/csrf_token.py

@db_injector
async def delete_expired_tokens(db: PrefectDBInterface, session: AsyncSession) -> int:
    """Delete expired CSRF tokens.

    Args:
        session (AsyncSession): The database session

    Returns:
        int: The number of tokens deleted
    """

    result = await session.execute(
        sa.delete(db.CsrfToken).where(
            db.CsrfToken.expiration < datetime.now(timezone.utc)
        )
    )
    return result.rowcount

read_token_for_client async

Read a CSRF token for a client.

Parameters:

Name	Type	Description	Default
session	AsyncSession	The database session	required
client	str	The client identifier	required

Returns:

Type	Description
Optional[CsrfToken]	Optional[core.CsrfToken]: The CSRF token, if it exists and is not expired.

Source code in prefect/server/models/csrf_token.py

@db_injector
async def read_token_for_client(
    db: PrefectDBInterface,
    session: AsyncSession,
    client: str,
) -> Optional[core.CsrfToken]:
    """Read a CSRF token for a client.

    Args:
        session (AsyncSession): The database session
        client (str): The client identifier

    Returns:
        Optional[core.CsrfToken]: The CSRF token, if it exists and is not
            expired.
    """
    token = (
        await session.execute(
            sa.select(db.CsrfToken).where(
                sa.and_(
                    db.CsrfToken.expiration > datetime.now(timezone.utc),
                    db.CsrfToken.client == client,
                )
            )
        )
    ).scalar_one_or_none()

    if token is None:
        return None

    return core.CsrfToken.from_orm(token)