routers.users

Endpoints for users.

View Source

  1"""Endpoints for users."""
  2
  3from fastapi import APIRouter, HTTPException, Response, UploadFile, status
  4from internal.auth.middleware import BearerAuthDep
  5from internal.auth.security import UpdatePasswordForm, update_pw
  6from internal.block.management import block_management
  7from internal.database.dependency import database_dependency
  8from internal.queries.admin_issue_reports import (
  9    AsyncQuerier as AdminIssueReportsQuerier,
 10)
 11from internal.queries.admin_issue_reports import CreateAdminIssueReportParams
 12from internal.queries.inbox import AsyncQuerier as InboxQuerier
 13from internal.queries.inbox import CreateInboxMessageParams
 14from internal.queries.models import (
 15    AdminIssueReport,
 16    AdminIssueType,
 17    Inbox,
 18    SellerIssueReport,
 19    SellerIssueType,
 20    UserRole,
 21)
 22from internal.queries.reservations import AsyncQuerier as ReservationsQuerier
 23from internal.queries.seller_issue_reports import (
 24    AsyncQuerier as SellerIssueReportsQuerier,
 25)
 26from internal.queries.seller_issue_reports import CreateSellerIssueReportParams
 27from internal.queries.user import AsyncQuerier as UserQuerier
 28from internal.queries.user import UpdateUserEmailParams
 29from pydantic import BaseModel, EmailStr
 30
 31router = APIRouter(prefix="/users", tags=["users"])
 32
 33
 34class SendMessageForm(BaseModel):
 35    """Form to send a new message."""
 36
 37    user_id: int
 38    message_subject: str
 39    message_text: str
 40
 41
 42@router.patch(
 43    "/me/password",
 44    status_code=status.HTTP_202_ACCEPTED,
 45    summary="Update password",
 46    description="Updates the password for the authenticated user.",
 47)
 48async def update_password(
 49    form: UpdatePasswordForm, conn: database_dependency, session: BearerAuthDep
 50) -> None:
 51    """Update users password.
 52
 53    Args:
 54      form: form for password change
 55      conn: database connection
 56      session: users session
 57    """
 58    await update_pw(session.email, form, conn)
 59
 60
 61@router.patch(
 62    "/me/email",
 63    status_code=status.HTTP_202_ACCEPTED,
 64    summary="Update email",
 65    description="Updates the email address for the authenticated user.",
 66)
 67async def update_email(
 68    email: EmailStr, conn: database_dependency, session: BearerAuthDep
 69) -> None:
 70    """Update users email.
 71
 72    Args:
 73      email: new users email
 74      conn: database connection
 75      session: users session
 76
 77    Raises:
 78      HTTPException: failed to update user email
 79    """
 80    user = await UserQuerier(conn).update_user_email(
 81        UpdateUserEmailParams(user_id=session.user_id, email=email)
 82    )
 83    if not user:
 84        raise HTTPException(
 85            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
 86            detail="Failed to update users email",
 87        )
 88
 89
 90@router.get(
 91    "/me/inbox",
 92    status_code=status.HTTP_200_OK,
 93    summary="Get user inbox",
 94    description="Gets all inbox messages for the authenticated user.",
 95    response_model=list[Inbox],
 96)
 97async def get_inbox(conn: database_dependency, session: BearerAuthDep) -> list[Inbox]:
 98    """Get inbox for user.
 99
100    Args:
101      conn: database connection
102      session: users session
103
104    Returns:
105      A list of inbox messages.
106    """
107    return [
108        msg async for msg in InboxQuerier(conn).get_user_inbox(user_id=session.user_id)
109    ]
110
111
112@router.post(
113    "/me/inbox",
114    status_code=status.HTTP_201_CREATED,
115    summary="Send an inbox message",
116    description="Sends a new message to a user.",
117    response_model=Inbox,
118)
119async def send_message(
120    form: SendMessageForm, conn: database_dependency, session: BearerAuthDep
121) -> Inbox:
122    """Send a message.
123
124    Args:
125      form: The message form
126      conn: database connection
127      session: users session
128
129    Returns:
130      The created message.
131
132    Raises:
133      HTTPException: if message sending fails
134    """
135    message = await InboxQuerier(conn).create_inbox_message(
136        CreateInboxMessageParams(
137            user_id=form.user_id,
138            sender_id=session.user_id,
139            message_subject=form.message_subject,
140            message_text=form.message_text,
141        )
142    )
143    if not message:
144        raise HTTPException(
145            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
146            detail="Failed to send message",
147        )
148    return message
149
150
151@router.get(
152    path="/id/{username}",
153    summary="Get user ID by username",
154    description="Retrieves a user ID and role by their username.",
155)
156async def get_user_id(username: str, conn: database_dependency) -> tuple[int, UserRole]:
157    """Get user id from username.
158
159    Args:
160        username: username
161        conn: database connection
162
163    Returns:
164        user id and role
165
166    Raises:
167        HTTPException: if failed to find user
168    """
169    if (user := await UserQuerier(conn).get_user_id(username=username)) is None:
170        raise HTTPException(status.HTTP_404_NOT_FOUND, "failed to find user")
171    return (user.user_id, user.role)
172
173
174@router.patch(
175    path="/me/image",
176    status_code=status.HTTP_202_ACCEPTED,
177    summary="Change profile image",
178    description="Updates the profile image for the authenticated user.",
179)
180async def change_profile_image(file: UploadFile, session: BearerAuthDep) -> None:
181    """Change user profile image.
182
183    Args:
184        file: profile image
185        session: user session
186    """
187    await block_management.upload_profile_image(session.user_id, file)
188
189
190@router.get(
191    path="/me/image",
192    status_code=status.HTTP_200_OK,
193    summary="Get own profile image",
194    description="Retrieves the profile image for the authenticated user.",
195)
196async def get_profile_image_me(session: BearerAuthDep) -> Response:
197    """Get own profile image.
198
199    Args:
200        session: user session
201
202    Returns:
203        user profile image
204    """
205    return Response(
206        block_management.get_profile_image(session.user_id), media_type="image/jpeg"
207    )
208
209
210@router.get(
211    path="/{user_id}/image",
212    status_code=status.HTTP_200_OK,
213    summary="Get user profile image",
214    description="Retrieves the profile image for a specific user by their ID.",
215)
216async def get_profile_image_by_id(user_id: int) -> Response:
217    """Get user profile image by ID.
218
219    Args:
220        user_id: user id
221
222    Returns:
223        user profile image
224    """
225    return Response(
226        block_management.get_profile_image(user_id), media_type="image/jpeg"
227    )
228
229
230class CreateSellerIssueReportForm(BaseModel):
231    """Seller issue report creation form."""
232
233    issue_type: SellerIssueType
234    description: str
235
236
237async def create_seller_issue_report(
238    reservation_id: int,
239    form: CreateSellerIssueReportForm,
240    conn: database_dependency,
241    user: BearerAuthDep,
242) -> SellerIssueReport:
243    """Create seller issue report.
244
245    Args:
246        reservation_id: reservation id from path
247        form: seller issue report creation form
248        conn: database connection
249        user: user session
250
251    Returns:
252        created seller issue report
253
254    Raises:
255        HTTPException: if failed to create report or reservation not owned by consumer
256    """
257    reservation = await ReservationsQuerier(conn).get_reservation(
258        reservation_id=reservation_id
259    )
260    if not reservation or reservation.consumer_id != user.user_id:
261        raise HTTPException(
262            status_code=status.HTTP_404_NOT_FOUND,
263            detail="Reservation not found or not owned by consumer",
264        )
265
266    report = await SellerIssueReportsQuerier(conn).create_seller_issue_report(
267        CreateSellerIssueReportParams(
268            reservation_id=reservation_id,
269            issue_type=form.issue_type,
270            description=form.description,
271        )
272    )
273    if not report:
274        raise HTTPException(
275            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
276            detail="Failed to create seller issue report",
277        )
278    return report
279
280
281class CreateAdminIssueReportForm(BaseModel):
282    """Admin issue report creation form."""
283
284    issue_type: AdminIssueType
285    description: str
286
287
288@router.post(
289    "/me/reports/admin",
290    status_code=status.HTTP_201_CREATED,
291    summary="Create admin issue report",
292    description="Creates a new admin issue report.",
293    tags=["reports"],
294)
295async def create_admin_issue_report(
296    form: CreateAdminIssueReportForm, conn: database_dependency, user: BearerAuthDep
297) -> AdminIssueReport:
298    """Create admin issue report.
299
300    Args:
301        form: admin issue report creation form
302        conn: database connection
303        user: user session
304
305    Returns:
306        created admin issue report
307
308    Raises:
309        HTTPException: if failed to create report
310    """
311    report = await AdminIssueReportsQuerier(conn).create_admin_issue_report(
312        CreateAdminIssueReportParams(
313            user_id=user.user_id,
314            issue_type=form.issue_type,
315            description=form.description,
316        )
317    )
318    if not report:
319        raise HTTPException(
320            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
321            detail="Failed to create admin issue report",
322        )
323    return report
324
325
326@router.get(
327    "/me/reports/admin",
328    status_code=status.HTTP_200_OK,
329    summary="Get admin issue reports",
330    description="Retrieves a list of admin issue reports created by the user.",
331    tags=["reports"],
332)
333async def get_admin_issue_reports(
334    conn: database_dependency, user: BearerAuthDep
335) -> list[AdminIssueReport]:
336    """Get admin issue reports.
337
338    Args:
339        conn: database connection
340        user: user session
341
342    Returns:
343        list of admin issue reports
344    """
345    return [
346        r
347        async for r in AdminIssueReportsQuerier(conn).get_admin_issue_reports_by_user(
348            user_id=user.user_id
349        )
350    ]
351
352
353@router.get(
354    "/me/reports/seller",
355    status_code=status.HTTP_200_OK,
356    summary="Get seller issue reports",
357    description="Retrieves a list of seller issue reports created by the user.",
358    tags=["reports"],
359)
360async def get_seller_issue_reports(
361    conn: database_dependency, user: BearerAuthDep
362) -> list[SellerIssueReport]:
363    """Get seller issue reports.
364
365    Args:
366        conn: database connection
367        user: user session
368
369    Returns:
370        list of seller issue reports
371    """
372    return [
373        r
374        async for r in SellerIssueReportsQuerier(conn).get_seller_issue_reports_by_user(
375            consumer_id=user.user_id
376        )
377    ]

router = <fastapi.routing.APIRouter object>

class SendMessageForm(pydantic.main.BaseModel): View Source

35class SendMessageForm(BaseModel):
36    """Form to send a new message."""
37
38    user_id: int
39    message_subject: str
40    message_text: str

Form to send a new message.

user_id: int = PydanticUndefined

message_subject: str = PydanticUndefined

message_text: str = PydanticUndefined

@router.patch('/me/password', status_code=status.HTTP_202_ACCEPTED, summary='Update password', description='Updates the password for the authenticated user.')

async def update_password( form: internal.auth.security.UpdatePasswordForm, conn: Annotated[sqlalchemy.ext.asyncio.engine.AsyncConnection, Depends(dependency=<bound method DatabaseManager.get_connection of <internal.database.manager.DatabaseManager object>>, use_cache=True, scope=None)], session: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function bearer_auth>, use_cache=True, scope=None, scopes=None)]) -> None: View Source

43@router.patch(
44    "/me/password",
45    status_code=status.HTTP_202_ACCEPTED,
46    summary="Update password",
47    description="Updates the password for the authenticated user.",
48)
49async def update_password(
50    form: UpdatePasswordForm, conn: database_dependency, session: BearerAuthDep
51) -> None:
52    """Update users password.
53
54    Args:
55      form: form for password change
56      conn: database connection
57      session: users session
58    """
59    await update_pw(session.email, form, conn)

Update users password.

Arguments:

form: form for password change
conn: database connection
session: users session

@router.patch('/me/email', status_code=status.HTTP_202_ACCEPTED, summary='Update email', description='Updates the email address for the authenticated user.')

async def update_email( email: pydantic.networks.EmailStr, conn: Annotated[sqlalchemy.ext.asyncio.engine.AsyncConnection, Depends(dependency=<bound method DatabaseManager.get_connection of <internal.database.manager.DatabaseManager object>>, use_cache=True, scope=None)], session: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function bearer_auth>, use_cache=True, scope=None, scopes=None)]) -> None: View Source

62@router.patch(
63    "/me/email",
64    status_code=status.HTTP_202_ACCEPTED,
65    summary="Update email",
66    description="Updates the email address for the authenticated user.",
67)
68async def update_email(
69    email: EmailStr, conn: database_dependency, session: BearerAuthDep
70) -> None:
71    """Update users email.
72
73    Args:
74      email: new users email
75      conn: database connection
76      session: users session
77
78    Raises:
79      HTTPException: failed to update user email
80    """
81    user = await UserQuerier(conn).update_user_email(
82        UpdateUserEmailParams(user_id=session.user_id, email=email)
83    )
84    if not user:
85        raise HTTPException(
86            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
87            detail="Failed to update users email",
88        )

Update users email.

Arguments:

email: new users email
conn: database connection
session: users session

Raises:

HTTPException: failed to update user email

@router.get('/me/inbox', status_code=status.HTTP_200_OK, summary='Get user inbox', description='Gets all inbox messages for the authenticated user.', response_model=list[Inbox])

async def get_inbox( conn: Annotated[sqlalchemy.ext.asyncio.engine.AsyncConnection, Depends(dependency=<bound method DatabaseManager.get_connection of <internal.database.manager.DatabaseManager object>>, use_cache=True, scope=None)], session: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function bearer_auth>, use_cache=True, scope=None, scopes=None)]) -> list[internal.queries.models.Inbox]: View Source

 91@router.get(
 92    "/me/inbox",
 93    status_code=status.HTTP_200_OK,
 94    summary="Get user inbox",
 95    description="Gets all inbox messages for the authenticated user.",
 96    response_model=list[Inbox],
 97)
 98async def get_inbox(conn: database_dependency, session: BearerAuthDep) -> list[Inbox]:
 99    """Get inbox for user.
100
101    Args:
102      conn: database connection
103      session: users session
104
105    Returns:
106      A list of inbox messages.
107    """
108    return [
109        msg async for msg in InboxQuerier(conn).get_user_inbox(user_id=session.user_id)
110    ]

Get inbox for user.

Arguments:

conn: database connection
session: users session

Returns:

A list of inbox messages.

@router.post('/me/inbox', status_code=status.HTTP_201_CREATED, summary='Send an inbox message', description='Sends a new message to a user.', response_model=Inbox)

async def send_message( form: SendMessageForm, conn: Annotated[sqlalchemy.ext.asyncio.engine.AsyncConnection, Depends(dependency=<bound method DatabaseManager.get_connection of <internal.database.manager.DatabaseManager object>>, use_cache=True, scope=None)], session: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function bearer_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.models.Inbox: View Source

113@router.post(
114    "/me/inbox",
115    status_code=status.HTTP_201_CREATED,
116    summary="Send an inbox message",
117    description="Sends a new message to a user.",
118    response_model=Inbox,
119)
120async def send_message(
121    form: SendMessageForm, conn: database_dependency, session: BearerAuthDep
122) -> Inbox:
123    """Send a message.
124
125    Args:
126      form: The message form
127      conn: database connection
128      session: users session
129
130    Returns:
131      The created message.
132
133    Raises:
134      HTTPException: if message sending fails
135    """
136    message = await InboxQuerier(conn).create_inbox_message(
137        CreateInboxMessageParams(
138            user_id=form.user_id,
139            sender_id=session.user_id,
140            message_subject=form.message_subject,
141            message_text=form.message_text,
142        )
143    )
144    if not message:
145        raise HTTPException(
146            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
147            detail="Failed to send message",
148        )
149    return message

Send a message.

Arguments:

form: The message form
conn: database connection
session: users session

Returns:

The created message.

Raises:

HTTPException: if message sending fails

@router.get(path='/id/{username}', summary='Get user ID by username', description='Retrieves a user ID and role by their username.')

async def get_user_id( username: str, conn: Annotated[sqlalchemy.ext.asyncio.engine.AsyncConnection, Depends(dependency=<bound method DatabaseManager.get_connection of <internal.database.manager.DatabaseManager object>>, use_cache=True, scope=None)]) -> tuple[int, internal.queries.models.UserRole]: View Source

152@router.get(
153    path="/id/{username}",
154    summary="Get user ID by username",
155    description="Retrieves a user ID and role by their username.",
156)
157async def get_user_id(username: str, conn: database_dependency) -> tuple[int, UserRole]:
158    """Get user id from username.
159
160    Args:
161        username: username
162        conn: database connection
163
164    Returns:
165        user id and role
166
167    Raises:
168        HTTPException: if failed to find user
169    """
170    if (user := await UserQuerier(conn).get_user_id(username=username)) is None:
171        raise HTTPException(status.HTTP_404_NOT_FOUND, "failed to find user")
172    return (user.user_id, user.role)

Get user id from username.

Arguments:

username: username
conn: database connection

Returns:

user id and role

Raises:

HTTPException: if failed to find user

@router.patch(path='/me/image', status_code=status.HTTP_202_ACCEPTED, summary='Change profile image', description='Updates the profile image for the authenticated user.')

async def change_profile_image( file: fastapi.datastructures.UploadFile, session: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function bearer_auth>, use_cache=True, scope=None, scopes=None)]) -> None: View Source

175@router.patch(
176    path="/me/image",
177    status_code=status.HTTP_202_ACCEPTED,
178    summary="Change profile image",
179    description="Updates the profile image for the authenticated user.",
180)
181async def change_profile_image(file: UploadFile, session: BearerAuthDep) -> None:
182    """Change user profile image.
183
184    Args:
185        file: profile image
186        session: user session
187    """
188    await block_management.upload_profile_image(session.user_id, file)

Change user profile image.

Arguments:

file: profile image
session: user session

@router.get(path='/me/image', status_code=status.HTTP_200_OK, summary='Get own profile image', description='Retrieves the profile image for the authenticated user.')

async def get_profile_image_me( session: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function bearer_auth>, use_cache=True, scope=None, scopes=None)]) -> starlette.responses.Response: View Source

191@router.get(
192    path="/me/image",
193    status_code=status.HTTP_200_OK,
194    summary="Get own profile image",
195    description="Retrieves the profile image for the authenticated user.",
196)
197async def get_profile_image_me(session: BearerAuthDep) -> Response:
198    """Get own profile image.
199
200    Args:
201        session: user session
202
203    Returns:
204        user profile image
205    """
206    return Response(
207        block_management.get_profile_image(session.user_id), media_type="image/jpeg"
208    )

Get own profile image.

Arguments:

session: user session

Returns:

user profile image

@router.get(path='/{user_id}/image', status_code=status.HTTP_200_OK, summary='Get user profile image', description='Retrieves the profile image for a specific user by their ID.')

async def get_profile_image_by_id(user_id: int) -> starlette.responses.Response: View Source

211@router.get(
212    path="/{user_id}/image",
213    status_code=status.HTTP_200_OK,
214    summary="Get user profile image",
215    description="Retrieves the profile image for a specific user by their ID.",
216)
217async def get_profile_image_by_id(user_id: int) -> Response:
218    """Get user profile image by ID.
219
220    Args:
221        user_id: user id
222
223    Returns:
224        user profile image
225    """
226    return Response(
227        block_management.get_profile_image(user_id), media_type="image/jpeg"
228    )

Get user profile image by ID.

Arguments:

user_id: user id

Returns:

user profile image

class CreateSellerIssueReportForm(pydantic.main.BaseModel): View Source

231class CreateSellerIssueReportForm(BaseModel):
232    """Seller issue report creation form."""
233
234    issue_type: SellerIssueType
235    description: str

Seller issue report creation form.

issue_type: internal.queries.models.SellerIssueType = PydanticUndefined

description: str = PydanticUndefined

async def create_seller_issue_report( reservation_id: int, form: CreateSellerIssueReportForm, conn: Annotated[sqlalchemy.ext.asyncio.engine.AsyncConnection, Depends(dependency=<bound method DatabaseManager.get_connection of <internal.database.manager.DatabaseManager object>>, use_cache=True, scope=None)], user: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function bearer_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.models.SellerIssueReport: View Source

238async def create_seller_issue_report(
239    reservation_id: int,
240    form: CreateSellerIssueReportForm,
241    conn: database_dependency,
242    user: BearerAuthDep,
243) -> SellerIssueReport:
244    """Create seller issue report.
245
246    Args:
247        reservation_id: reservation id from path
248        form: seller issue report creation form
249        conn: database connection
250        user: user session
251
252    Returns:
253        created seller issue report
254
255    Raises:
256        HTTPException: if failed to create report or reservation not owned by consumer
257    """
258    reservation = await ReservationsQuerier(conn).get_reservation(
259        reservation_id=reservation_id
260    )
261    if not reservation or reservation.consumer_id != user.user_id:
262        raise HTTPException(
263            status_code=status.HTTP_404_NOT_FOUND,
264            detail="Reservation not found or not owned by consumer",
265        )
266
267    report = await SellerIssueReportsQuerier(conn).create_seller_issue_report(
268        CreateSellerIssueReportParams(
269            reservation_id=reservation_id,
270            issue_type=form.issue_type,
271            description=form.description,
272        )
273    )
274    if not report:
275        raise HTTPException(
276            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
277            detail="Failed to create seller issue report",
278        )
279    return report

Create seller issue report.

Arguments:

reservation_id: reservation id from path
form: seller issue report creation form
conn: database connection
user: user session

Returns:

created seller issue report

Raises:

HTTPException: if failed to create report or reservation not owned by consumer

class CreateAdminIssueReportForm(pydantic.main.BaseModel): View Source

282class CreateAdminIssueReportForm(BaseModel):
283    """Admin issue report creation form."""
284
285    issue_type: AdminIssueType
286    description: str

Admin issue report creation form.

issue_type: internal.queries.models.AdminIssueType = PydanticUndefined

description: str = PydanticUndefined

@router.post('/me/reports/admin', status_code=status.HTTP_201_CREATED, summary='Create admin issue report', description='Creates a new admin issue report.', tags=['reports'])

async def create_admin_issue_report( form: CreateAdminIssueReportForm, conn: Annotated[sqlalchemy.ext.asyncio.engine.AsyncConnection, Depends(dependency=<bound method DatabaseManager.get_connection of <internal.database.manager.DatabaseManager object>>, use_cache=True, scope=None)], user: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function bearer_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.models.AdminIssueReport: View Source

289@router.post(
290    "/me/reports/admin",
291    status_code=status.HTTP_201_CREATED,
292    summary="Create admin issue report",
293    description="Creates a new admin issue report.",
294    tags=["reports"],
295)
296async def create_admin_issue_report(
297    form: CreateAdminIssueReportForm, conn: database_dependency, user: BearerAuthDep
298) -> AdminIssueReport:
299    """Create admin issue report.
300
301    Args:
302        form: admin issue report creation form
303        conn: database connection
304        user: user session
305
306    Returns:
307        created admin issue report
308
309    Raises:
310        HTTPException: if failed to create report
311    """
312    report = await AdminIssueReportsQuerier(conn).create_admin_issue_report(
313        CreateAdminIssueReportParams(
314            user_id=user.user_id,
315            issue_type=form.issue_type,
316            description=form.description,
317        )
318    )
319    if not report:
320        raise HTTPException(
321            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
322            detail="Failed to create admin issue report",
323        )
324    return report

Create admin issue report.

Arguments:

form: admin issue report creation form
conn: database connection
user: user session

Returns:

created admin issue report

Raises:

HTTPException: if failed to create report

@router.get('/me/reports/admin', status_code=status.HTTP_200_OK, summary='Get admin issue reports', description='Retrieves a list of admin issue reports created by the user.', tags=['reports'])

async def get_admin_issue_reports( conn: Annotated[sqlalchemy.ext.asyncio.engine.AsyncConnection, Depends(dependency=<bound method DatabaseManager.get_connection of <internal.database.manager.DatabaseManager object>>, use_cache=True, scope=None)], user: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function bearer_auth>, use_cache=True, scope=None, scopes=None)]) -> list[internal.queries.models.AdminIssueReport]: View Source

327@router.get(
328    "/me/reports/admin",
329    status_code=status.HTTP_200_OK,
330    summary="Get admin issue reports",
331    description="Retrieves a list of admin issue reports created by the user.",
332    tags=["reports"],
333)
334async def get_admin_issue_reports(
335    conn: database_dependency, user: BearerAuthDep
336) -> list[AdminIssueReport]:
337    """Get admin issue reports.
338
339    Args:
340        conn: database connection
341        user: user session
342
343    Returns:
344        list of admin issue reports
345    """
346    return [
347        r
348        async for r in AdminIssueReportsQuerier(conn).get_admin_issue_reports_by_user(
349            user_id=user.user_id
350        )
351    ]

Get admin issue reports.

Arguments:

conn: database connection
user: user session

Returns:

list of admin issue reports

@router.get('/me/reports/seller', status_code=status.HTTP_200_OK, summary='Get seller issue reports', description='Retrieves a list of seller issue reports created by the user.', tags=['reports'])

async def get_seller_issue_reports( conn: Annotated[sqlalchemy.ext.asyncio.engine.AsyncConnection, Depends(dependency=<bound method DatabaseManager.get_connection of <internal.database.manager.DatabaseManager object>>, use_cache=True, scope=None)], user: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function bearer_auth>, use_cache=True, scope=None, scopes=None)]) -> list[internal.queries.models.SellerIssueReport]: View Source

354@router.get(
355    "/me/reports/seller",
356    status_code=status.HTTP_200_OK,
357    summary="Get seller issue reports",
358    description="Retrieves a list of seller issue reports created by the user.",
359    tags=["reports"],
360)
361async def get_seller_issue_reports(
362    conn: database_dependency, user: BearerAuthDep
363) -> list[SellerIssueReport]:
364    """Get seller issue reports.
365
366    Args:
367        conn: database connection
368        user: user session
369
370    Returns:
371        list of seller issue reports
372    """
373    return [
374        r
375        async for r in SellerIssueReportsQuerier(conn).get_seller_issue_reports_by_user(
376            consumer_id=user.user_id
377        )
378    ]

Get seller issue reports.

Arguments:

conn: database connection
user: user session

Returns:

list of seller issue reports