feat: Implement admin user management endpoints and user update schema

This commit is contained in:
JSC
2025-08-09 22:37:51 +02:00
parent 734521c5c3
commit 9e07ce393f
6 changed files with 220 additions and 3 deletions

View File

@@ -4,6 +4,7 @@ from typing import Any
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.orm import selectinload
from app.core.logging import get_logger
from app.models.plan import Plan
@@ -20,6 +21,42 @@ class UserRepository(BaseRepository[User]):
"""Initialize the user repository."""
super().__init__(User, session)
async def get_all_with_plan(
self,
limit: int = 100,
offset: int = 0,
) -> list[User]:
"""Get all users with plan relationship loaded."""
try:
statement = (
select(User)
.options(selectinload(User.plan))
.limit(limit)
.offset(offset)
)
result = await self.session.exec(statement)
return list(result.all())
except Exception:
logger.exception("Failed to get all users with plan")
raise
async def get_by_id_with_plan(self, entity_id: int) -> User | None:
"""Get a user by ID with plan relationship loaded."""
try:
statement = (
select(User)
.options(selectinload(User.plan))
.where(User.id == entity_id)
)
result = await self.session.exec(statement)
return result.first()
except Exception:
logger.exception(
"Failed to get user by ID with plan: %s",
entity_id,
)
raise
async def get_by_email(self, email: str) -> User | None:
"""Get a user by email address."""
try: