service_man/api/infrastructure/persistence/repositories/user_repository.py

46 lines
1.4 KiB
Python
Raw Normal View History

2024-04-01 12:19:10 +03:00
from sqlalchemy import text
2024-03-31 04:18:41 +03:00
from sqlalchemy.ext.asyncio import AsyncSession
from api.domain.user import User, UserRepository
2024-04-04 13:51:51 +03:00
from api.domain.user.model import UserEmail, UserFirstName, UserId
2024-03-31 04:18:41 +03:00
class SqlAlchemyUserRepository(UserRepository):
def __init__(self, session: AsyncSession) -> None:
self.session = session
async def create_user(self, user: User) -> None:
2024-04-01 12:19:10 +03:00
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,
},
2024-03-31 21:07:59 +03:00
)
2024-03-31 04:18:41 +03:00
async def get_user(self, filter: dict) -> User | None:
2024-04-04 13:51:51 +03:00
stmt = text("""SELECT * FROM users WHERE email = :val""")
result = await self.session.execute(stmt, {"val": filter["email"]})
2024-04-08 00:31:15 +03:00
result = result.mappings().one_or_none()
if result is None:
2024-04-04 13:51:51 +03:00
return None
2024-04-08 00:31:15 +03:00
2024-04-04 13:51:51 +03:00
return User(
id=UserId(result.id),
name=UserFirstName(result.name),
email=UserEmail(result.email),
hashed_password=result.hashed_password,
)
2024-03-31 04:18:41 +03:00
async def get_users(self) -> list[User]:
return []