internal.auth.token

Manages tokens in database.

 1"""Manages tokens in database."""
 2
 3from datetime import UTC, datetime, timedelta
 4
 5from sqlalchemy.ext.asyncio import AsyncConnection
 6
 7from internal.queries.models import Token
 8from internal.queries.token import AsyncQuerier as TokenQuerier
 9from internal.queries.token import CreateTokenParams
10from internal.settings.env import auth_settings
11
12from .security import generate_token
13
14
15async def create_token(user_id: int, conn: AsyncConnection) -> Token:
16    """Create, insert into database and return token for the user.
17
18    Args:
19      user_id: user id for which to create token
20      conn: database connection
21
22    Returns:
23      row from the database for the created token
24
25    Raises:
26      ValueError: if database failed to insert token
27    """
28    token = await TokenQuerier(conn).create_token(
29        CreateTokenParams(
30            user_id=user_id,
31            token=generate_token(),
32            expires_at=datetime.now(UTC)
33            + timedelta(seconds=auth_settings.token_exparation),
34        )
35    )
36    if not token:
37        raise ValueError("Failed to create token")
38    return token
39
40
41async def delete_token(token: str, conn: AsyncConnection) -> None:
42    """Delete given token from database.
43
44    Args:
45      token: token to delete
46      conn: database connection
47
48    Raises:
49      ValueError: if failed to delete or find given token
50    """
51    deleted_token = await TokenQuerier(conn).delete_token(token=token)
52    if not deleted_token:
53        raise ValueError("Failed to delete token")
async def create_token( user_id: int, conn: sqlalchemy.ext.asyncio.engine.AsyncConnection) -> internal.queries.models.Token: 
16async def create_token(user_id: int, conn: AsyncConnection) -> Token:
17    """Create, insert into database and return token for the user.
18
19    Args:
20      user_id: user id for which to create token
21      conn: database connection
22
23    Returns:
24      row from the database for the created token
25
26    Raises:
27      ValueError: if database failed to insert token
28    """
29    token = await TokenQuerier(conn).create_token(
30        CreateTokenParams(
31            user_id=user_id,
32            token=generate_token(),
33            expires_at=datetime.now(UTC)
34            + timedelta(seconds=auth_settings.token_exparation),
35        )
36    )
37    if not token:
38        raise ValueError("Failed to create token")
39    return token

Create, insert into database and return token for the user.

Arguments:
  • user_id: user id for which to create token
  • conn: database connection
Returns:

row from the database for the created token

Raises:
  • ValueError: if database failed to insert token
async def delete_token(token: str, conn: sqlalchemy.ext.asyncio.engine.AsyncConnection) -> None: 
42async def delete_token(token: str, conn: AsyncConnection) -> None:
43    """Delete given token from database.
44
45    Args:
46      token: token to delete
47      conn: database connection
48
49    Raises:
50      ValueError: if failed to delete or find given token
51    """
52    deleted_token = await TokenQuerier(conn).delete_token(token=token)
53    if not deleted_token:
54        raise ValueError("Failed to delete token")

Delete given token from database.

Arguments:
  • token: token to delete
  • conn: database connection
Raises:
  • ValueError: if failed to delete or find given token