42 lines
941 B
Python
42 lines
941 B
Python
from datetime import datetime
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from db.models.link_codes import LinkCode
|
|
from schemas.enums import LinkCodeStatus
|
|
|
|
|
|
async def create_link_code(
|
|
session: AsyncSession,
|
|
*,
|
|
code: str,
|
|
user_id: int,
|
|
status: LinkCodeStatus,
|
|
expires_at: datetime,
|
|
) -> LinkCode:
|
|
link_code = LinkCode(
|
|
code=code,
|
|
user_id=user_id,
|
|
status=status,
|
|
expires_at=expires_at,
|
|
)
|
|
|
|
session.add(link_code)
|
|
await session.commit()
|
|
return link_code
|
|
|
|
|
|
async def get_link_code_by_code(session: AsyncSession, code: str) -> LinkCode | None:
|
|
stmt = select(LinkCode).where(LinkCode.code == code)
|
|
r = await session.execute(stmt)
|
|
|
|
return r.scalar_one_or_none()
|
|
|
|
|
|
async def use_link_code(session: AsyncSession, code: LinkCode) -> LinkCode:
|
|
code.status = LinkCodeStatus.USED
|
|
await session.commit()
|
|
|
|
return code
|