internal.auth.creation

  1"""Creates seller and consumer entities."""
  2
  3from pydantic import BaseModel, EmailStr, SecretStr
  4from sqlalchemy.ext.asyncio import AsyncConnection
  5
  6from internal.queries.admin import AsyncQuerier as AdminQuerier
  7from internal.queries.admin import CreateAdminParams
  8from internal.queries.consumer import AsyncQuerier as ConsumerQuerier
  9from internal.queries.consumer import CreateConsumerParams
 10from internal.queries.models import Admin, Consumer, Seller, UserRole
 11from internal.queries.seller import AsyncQuerier as SellerQuerier
 12from internal.queries.seller import CreateSellerParams
 13from internal.queries.user import AsyncQuerier as UserQuerier
 14from internal.queries.user import CreateUserParams, CreateUserRow
 15
 16from .security import hash_password
 17
 18
 19async def create_user(
 20    username: str,
 21    email: EmailStr,
 22    password: SecretStr,
 23    role: UserRole,
 24    conn: AsyncConnection,
 25) -> CreateUserRow:
 26    """Create and insert base user entry.
 27
 28    Args:
 29      username: username for user
 30      email: users email
 31      password: users plain text password
 32      role: users role, seller or consumer
 33      conn: database connection
 34
 35    Returns:
 36      row from database with created users information
 37
 38    Raises:
 39      ValueError: if database failed to create user
 40    """
 41    pw_hash = hash_password(password.get_secret_value())
 42    user = await UserQuerier(conn).create_user(
 43        CreateUserParams(username=username, email=email, pw_hash=pw_hash, role=role)
 44    )
 45    if not user:
 46        raise ValueError("Failed to create user")
 47    return user
 48
 49
 50class CreateConsumerForm(BaseModel):
 51    """Form with information to create consumer."""
 52
 53    username: str
 54    email: EmailStr
 55    password: SecretStr
 56    first_name: str
 57    last_name: str
 58
 59
 60async def create_consumer(form: CreateConsumerForm, conn: AsyncConnection) -> Consumer:
 61    """Create and insert consumer and user entry.
 62
 63    Args:
 64      form: form with consumers information
 65      conn: database connection
 66
 67    Returns:
 68      row with created consumers information from the database
 69
 70    Raises:
 71      ValueError: if database failed to create consumer
 72    """
 73    user = await create_user(
 74        form.username, form.email, form.password, UserRole.CONSUMER, conn
 75    )
 76    consumer = await ConsumerQuerier(conn).create_consumer(
 77        CreateConsumerParams(
 78            user_id=user.user_id, fname=form.first_name, lname=form.last_name
 79        )
 80    )
 81    if not consumer:
 82        raise ValueError("Failed to create consumer")
 83    return consumer
 84
 85
 86class CreateSellerForm(BaseModel):
 87    """Form with information to create seller."""
 88
 89    username: str
 90    email: EmailStr
 91    password: SecretStr
 92    seller_name: str
 93    address_line1: str
 94    address_line2: str | None = None
 95    city: str
 96    post_code: str
 97    region: str | None = None
 98    country: str
 99    latitude: float
100    longitude: float
101
102
103async def create_seller(form: CreateSellerForm, conn: AsyncConnection) -> Seller:
104    """Creates and inserts seller and user entry.
105
106    Args:
107      form: form with information for seller creation
108      conn: database connection
109
110    Returns:
111     row with created sellers information from the database
112
113    Raises:
114      ValueError: database failed to create a seller
115    """
116    user = await create_user(
117        form.username, form.email, form.password, UserRole.SELLER, conn
118    )
119    seller = await SellerQuerier(conn).create_seller(
120        CreateSellerParams(
121            user_id=user.user_id,
122            seller_name=form.seller_name,
123            address_line1=form.address_line1,
124            address_line2=form.address_line2,
125            city=form.city,
126            post_code=form.post_code,
127            region=form.region,
128            country=form.country,
129            latitude=form.latitude,
130            longitude=form.longitude,
131        )
132    )
133    if not seller:
134        raise ValueError("Failed to create seller")
135    return seller
136
137
138class CreateAdminForm(BaseModel):
139    """Form with information to create admin."""
140
141    username: str
142    email: EmailStr
143    password: SecretStr
144    first_name: str
145    last_name: str
146
147
148async def create_admin(form: CreateAdminForm, conn: AsyncConnection) -> Admin:
149    """Create and insert admin and user entry.
150
151    Args:
152      form: form with admin information
153      conn: database connection
154
155    Returns:
156      row with created admin information from the database
157
158    Raises:
159      ValueError: if database failed to create admin
160    """
161    user = await create_user(
162        form.username, form.email, form.password, UserRole.ADMIN, conn
163    )
164    admin = await AdminQuerier(conn).create_admin(
165        CreateAdminParams(
166            user_id=user.user_id, fname=form.first_name, lname=form.last_name
167        )
168    )
169    if not admin:
170        raise ValueError("Failed to create admin")
171    return admin