46 lines
1.5 KiB
Python
46 lines
1.5 KiB
Python
# from sqlalchemy import text
|
|
# from sqlalchemy.ext.asyncio import AsyncSession
|
|
#
|
|
# from api.domain.company import CompanyRepository, company
|
|
# from api.domain.user.model import UserEmail, UserFirstName, UserId
|
|
#
|
|
#
|
|
# class SqlAlchemyUserRepository(UserRepository):
|
|
# def __init__(self, session: AsyncSession) -> None:
|
|
# self.session = session
|
|
#
|
|
# async def create_user(self, user: User) -> None:
|
|
# stmt = text(
|
|
# """INSERT INTO users (id, name, email, hashed_password)
|
|
# VALUES(:id, :name, :email, :hashed_password)
|
|
# """
|
|
# )
|
|
# await self.session.execute(
|
|
# stmt,
|
|
# {
|
|
# "id": str(user.id.value),
|
|
# "name": user.name.value,
|
|
# "email": user.email.value,
|
|
# "hashed_password": user.hashed_password,
|
|
# },
|
|
# )
|
|
#
|
|
# async def get_user(self, filter: dict) -> User | None:
|
|
# stmt = text("""SELECT * FROM users WHERE email = :val""")
|
|
# result = await self.session.execute(stmt, {"val": filter["email"]})
|
|
#
|
|
# result = result.mappings().one_or_none()
|
|
#
|
|
# if result is None:
|
|
# return None
|
|
#
|
|
# return User(
|
|
# id=UserId(result.id),
|
|
# name=UserFirstName(result.name),
|
|
# email=UserEmail(result.email),
|
|
# hashed_password=result.hashed_password,
|
|
# )
|
|
#
|
|
# async def get_users(self) -> list[User]:
|
|
# return []
|