routers.admins

Endpoints for admins.

   1"""Endpoints for admins."""
   2
   3from fastapi import APIRouter, HTTPException, status
   4from internal.auth.creation import CreateAdminForm, create_admin
   5from internal.auth.middleware import AdminAuthDep, RootAuthDep
   6from internal.auth.security import hash_password
   7from internal.database.dependency import database_dependency
   8from internal.queries.admin import AsyncQuerier as AdminQuerier
   9from internal.queries.admin import (
  10    GetAdminRow,
  11    GetAdminsRow,
  12    SetIsAdminActiveParams,
  13    UpdateAdminParams,
  14)
  15from internal.queries.admin_issue_reports import (
  16    AsyncQuerier as AdminIssueReportsQuerier,
  17)
  18from internal.queries.admin_issue_reports import UpdateAdminIssueReportStatusParams
  19from internal.queries.allergens import AsyncQuerier as AllergensQuerier
  20from internal.queries.allergens import UpdateAllergenParams
  21from internal.queries.badge import AsyncQuerier as BadgeQuerier
  22from internal.queries.badge import UpdateBadgeParams
  23from internal.queries.bundle import AsyncQuerier as BundleQuerier
  24from internal.queries.category import AsyncQuerier as CategoryQuerier
  25from internal.queries.category import CreateCategoryParams, UpdateCategoryParams
  26from internal.queries.consumer import AsyncQuerier as ConsumerQuerier
  27from internal.queries.consumer import GetConsumersRow, UpdateConsumerParams
  28from internal.queries.inbox import AsyncQuerier as InboxQuerier
  29from internal.queries.inbox import CreateInboxMessageParams
  30from internal.queries.models import (
  31    Admin,
  32    AdminIssueReport,
  33    Allergen,
  34    Badge,
  35    Bundle,
  36    Category,
  37    Consumer,
  38    Inbox,
  39    IssueStatus,
  40    Reservation,
  41    Seller,
  42    SellerIssueReport,
  43)
  44from internal.queries.reservations import AsyncQuerier as ReservationsQuerier
  45from internal.queries.seller import AsyncQuerier as SellerQuerier
  46from internal.queries.seller import (
  47    GetSellersRow,
  48    UpdateSellerParams,
  49    VerifySellerParams,
  50)
  51from internal.queries.seller_issue_reports import (
  52    AsyncQuerier as SellerIssueReportsQuerier,
  53)
  54from internal.queries.seller_issue_reports import UpdateSellerIssueReportStatusParams
  55from internal.queries.user import AsyncQuerier as UserQuerier
  56from internal.queries.user import (
  57    DeleteUserRow,
  58    GetUsersRow,
  59    UpdateUserEmailParams,
  60    UpdateUserEmailRow,
  61    UpdateUserPasswordParams,
  62    UpdateUserPasswordRow,
  63)
  64from pydantic import BaseModel, SecretStr
  65
  66router = APIRouter(prefix="/admins", tags=["admins"])
  67
  68
  69class UpdateAdminForm(BaseModel):
  70    """Admin name update form."""
  71
  72    first_name: str
  73    last_name: str
  74
  75
  76@router.post(
  77    "",
  78    status_code=status.HTTP_201_CREATED,
  79    summary="Create admin",
  80    description="Create admin by root user",
  81    tags=["root admin"],
  82)
  83async def register_admin(
  84    form: CreateAdminForm, conn: database_dependency, _: RootAuthDep
  85) -> None:
  86    """Create admin by root user.
  87
  88    Args:
  89        form: new admin information
  90        conn: database connection
  91    """
  92    await create_admin(form, conn)
  93
  94
  95@router.get(
  96    "",
  97    status_code=status.HTTP_200_OK,
  98    summary="Get all admins",
  99    description="Retrieves a list of all registered admins by root user.",
 100    tags=["root admin"],
 101)
 102async def get_admins(conn: database_dependency, _: RootAuthDep) -> list[GetAdminsRow]:
 103    """Get all admins by root user.
 104
 105    Args:
 106        conn: database connection
 107
 108    Returns:
 109        list of all admins
 110    """
 111    return [admin_row async for admin_row in AdminQuerier(conn).get_admins()]
 112
 113
 114@router.get(
 115    "/me",
 116    status_code=status.HTTP_200_OK,
 117    summary="Get authenticated admin",
 118    description="Retrieves the profile of the authenticated admin.",
 119)
 120async def get_admin_me(
 121    conn: database_dependency, admin_session: AdminAuthDep
 122) -> GetAdminRow:
 123    """Get authenticated admin profile.
 124
 125    Args:
 126        conn: database connection
 127        admin_session: admin session
 128
 129    Returns:
 130        admin profile
 131
 132    Raises:
 133        HTTPException: if admin not found
 134    """
 135    admin_profile = await AdminQuerier(conn).get_admin(user_id=admin_session.user_id)
 136    if not admin_profile:
 137        raise HTTPException(
 138            status_code=status.HTTP_404_NOT_FOUND, detail="Admin profile not found"
 139        )
 140    return admin_profile
 141
 142
 143@router.get(
 144    "/{admin_id}",
 145    status_code=status.HTTP_200_OK,
 146    summary="Get admin by ID",
 147    description="Retrieves the profile of an admin by their unique ID by root user.",
 148    tags=["root admin"],
 149)
 150async def get_admin_by_id(
 151    admin_id: int, conn: database_dependency, _: RootAuthDep
 152) -> GetAdminRow:
 153    """Get admin profile by ID by root user.
 154
 155    Args:
 156        admin_id: unique identifier of the admin
 157        conn: database connection
 158
 159    Returns:
 160        admin profile
 161
 162    Raises:
 163        HTTPException: if admin not found
 164    """
 165    admin_profile = await AdminQuerier(conn).get_admin(user_id=admin_id)
 166    if not admin_profile:
 167        raise HTTPException(
 168            status_code=status.HTTP_404_NOT_FOUND, detail="Admin not found"
 169        )
 170    return admin_profile
 171
 172
 173@router.patch(
 174    "/{admin_id}",
 175    status_code=status.HTTP_200_OK,
 176    summary="Update admin profile",
 177    description="Updates the profile information for an admin by root user.",
 178    tags=["root admin"],
 179)
 180async def update_admin(
 181    admin_id: int, form: UpdateAdminForm, conn: database_dependency, _: RootAuthDep
 182) -> Admin:
 183    """Admin name update by root user.
 184
 185    Args:
 186        admin_id: admin id
 187        form: admin update form
 188        conn: database connection
 189
 190    Returns:
 191        updated admin
 192
 193    Raises:
 194        HTTPException: if failed to update admin
 195    """
 196    updated_admin_profile = await AdminQuerier(conn).update_admin(
 197        UpdateAdminParams(user_id=admin_id, fname=form.first_name, lname=form.last_name)
 198    )
 199    if not updated_admin_profile:
 200        raise HTTPException(
 201            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
 202            detail="Failed to update admin",
 203        )
 204    return updated_admin_profile
 205
 206
 207@router.patch(
 208    "/{admin_id}/deactivate",
 209    status_code=status.HTTP_200_OK,
 210    summary="Deactivate admin",
 211    description="Deactivate admin by root user",
 212    tags=["root admin"],
 213)
 214async def deactivate_admin(
 215    admin_id: int, conn: database_dependency, _: RootAuthDep
 216) -> Admin:
 217    """Deactivate admin.
 218
 219    Args:
 220        admin_id: admin id
 221        conn: database connection
 222
 223    Returns:
 224        deactivated admin
 225
 226    Raises:
 227        HTTPException: if failed to find admin
 228    """
 229    admin_deactivation_result = await AdminQuerier(conn).set_is_admin_active(
 230        SetIsAdminActiveParams(user_id=admin_id, active=False)
 231    )
 232    if not admin_deactivation_result:
 233        raise HTTPException(status.HTTP_400_BAD_REQUEST, "No admin was found")
 234    return admin_deactivation_result
 235
 236
 237@router.patch(
 238    "/{admin_id}/activate",
 239    status_code=status.HTTP_200_OK,
 240    summary="Activate admin",
 241    description="Activate admin by root user",
 242    tags=["root admin"],
 243)
 244async def activate_admin(
 245    admin_id: int, conn: database_dependency, _: RootAuthDep
 246) -> Admin:
 247    """Activate admin.
 248
 249    Args:
 250        admin_id: admin id
 251        conn: database connection
 252
 253    Returns:
 254        activated admin
 255
 256    Raises:
 257        HTTPException: if failed to find admin
 258    """
 259    admin_activation_result = await AdminQuerier(conn).set_is_admin_active(
 260        SetIsAdminActiveParams(user_id=admin_id, active=True)
 261    )
 262    if not admin_activation_result:
 263        raise HTTPException(status.HTTP_400_BAD_REQUEST, "No admin was found")
 264    return admin_activation_result
 265
 266
 267@router.get(
 268    "/database/users",
 269    status_code=status.HTTP_200_OK,
 270    summary="Get all users",
 271    description="Retrieves a list of all users in the system.",
 272)
 273async def get_all_users(
 274    conn: database_dependency, _: AdminAuthDep
 275) -> list[GetUsersRow]:
 276    """Get all users.
 277
 278    Args:
 279        conn: database connection
 280
 281    Returns:
 282        list of all users
 283    """
 284    return [user_row async for user_row in UserQuerier(conn).get_users()]
 285
 286
 287class UpdateUserEmailForm(BaseModel):
 288    """User email update form."""
 289
 290    email: str
 291
 292
 293@router.patch(
 294    "/database/users/{user_id}/email",
 295    status_code=status.HTTP_200_OK,
 296    summary="Update user email",
 297    description="Updates the email address for a specific user.",
 298)
 299async def update_user_email(
 300    user_id: int, form: UpdateUserEmailForm, conn: database_dependency, _: AdminAuthDep
 301) -> UpdateUserEmailRow:
 302    """Update user email.
 303
 304    Args:
 305        user_id: user id
 306        form: email update form
 307        conn: database connection
 308
 309    Returns:
 310        updated user record
 311
 312    Raises:
 313        HTTPException: if failed to update email
 314    """
 315    updated_user_email = await UserQuerier(conn).update_user_email(
 316        UpdateUserEmailParams(user_id=user_id, email=form.email)
 317    )
 318    if not updated_user_email:
 319        raise HTTPException(status.HTTP_404_NOT_FOUND, "User not found")
 320    return updated_user_email
 321
 322
 323class UpdateUserPasswordForm(BaseModel):
 324    """User password update form."""
 325
 326    password: SecretStr
 327
 328
 329@router.patch(
 330    "/database/users/{user_id}/password",
 331    status_code=status.HTTP_200_OK,
 332    summary="Update user password",
 333    description="Updates the password for a specific user.",
 334)
 335async def update_user_password(
 336    user_id: int,
 337    form: UpdateUserPasswordForm,
 338    conn: database_dependency,
 339    _: AdminAuthDep,
 340) -> UpdateUserPasswordRow:
 341    """Update user password.
 342
 343    Args:
 344        user_id: user id
 345        form: password update form
 346        conn: database connection
 347
 348    Returns:
 349        updated user record
 350
 351    Raises:
 352        HTTPException: if failed to update password
 353    """
 354    hashed_pw = hash_password(form.password.get_secret_value())
 355    updated_user_password = await UserQuerier(conn).update_user_password(
 356        UpdateUserPasswordParams(user_id=user_id, pw_hash=hashed_pw)
 357    )
 358    if not updated_user_password:
 359        raise HTTPException(status.HTTP_404_NOT_FOUND, "User not found")
 360    return updated_user_password
 361
 362
 363@router.delete(
 364    "/database/users/{user_id}",
 365    status_code=status.HTTP_200_OK,
 366    summary="Delete user",
 367    description="Deletes a specific user from the system.",
 368)
 369async def delete_user(
 370    user_id: int, conn: database_dependency, _: AdminAuthDep
 371) -> DeleteUserRow:
 372    """Delete user.
 373
 374    Args:
 375        user_id: user id
 376        conn: database connection
 377
 378    Returns:
 379        deleted user
 380
 381    Raises:
 382        HTTPException: if user not found
 383    """
 384    deleted_user = await UserQuerier(conn).delete_user(user_id=user_id)
 385    if not deleted_user:
 386        raise HTTPException(status.HTTP_404_NOT_FOUND, "User not found")
 387    return deleted_user
 388
 389
 390@router.get(
 391    "/database/sellers",
 392    status_code=status.HTTP_200_OK,
 393    summary="Get all sellers",
 394    description="Retrieves a list of all sellers in the system.",
 395)
 396async def get_all_sellers(
 397    conn: database_dependency, _: AdminAuthDep
 398) -> list[GetSellersRow]:
 399    """Get all sellers.
 400
 401    Args:
 402        conn: database connection
 403
 404    Returns:
 405        list of all sellers
 406    """
 407    return [seller_row async for seller_row in SellerQuerier(conn).get_sellers()]
 408
 409
 410@router.patch(
 411    "/database/sellers/{seller_id}/verify",
 412    status_code=status.HTTP_200_OK,
 413    summary="Verify seller",
 414    description="Verifies a seller. Only possible if coordinates exist.",
 415)
 416async def verify_seller(
 417    seller_id: int, conn: database_dependency, admin_session: AdminAuthDep
 418) -> Seller:
 419    """Verify seller. Only possible if coordinates exist.
 420
 421    Args:
 422        seller_id: seller id
 423        conn: database connection
 424        admin_session: admin session
 425
 426    Returns:
 427        verified seller
 428
 429    Raises:
 430        HTTPException: if seller not found or coordinates missing
 431    """
 432    seller_querier = SellerQuerier(conn)
 433    seller_profile = await seller_querier.get_seller(user_id=seller_id)
 434    if not seller_profile:
 435        raise HTTPException(status.HTTP_404_NOT_FOUND, "Seller not found")
 436
 437    if seller_profile.latitude is None or seller_profile.longitude is None:
 438        raise HTTPException(
 439            status_code=status.HTTP_400_BAD_REQUEST,
 440            detail="Cannot verify seller without valid coordinates",
 441        )
 442
 443    verified_seller_profile = await seller_querier.verify_seller(
 444        VerifySellerParams(user_id=seller_id, verified_by=admin_session.user_id)
 445    )
 446    if not verified_seller_profile:
 447        raise HTTPException(status.HTTP_404_NOT_FOUND, "Seller not found")
 448    return verified_seller_profile
 449
 450
 451@router.patch(
 452    "/database/sellers/{seller_id}/unverify",
 453    status_code=status.HTTP_200_OK,
 454    summary="Unverify seller",
 455    description="Removes verification status from a seller.",
 456)
 457async def unverify_seller(
 458    seller_id: int, conn: database_dependency, _: AdminAuthDep
 459) -> Seller:
 460    """Unverify seller.
 461
 462    Args:
 463        seller_id: seller id
 464        conn: database connection
 465
 466    Returns:
 467        unverified seller
 468
 469    Raises:
 470        HTTPException: if seller not found
 471    """
 472    unverified_seller_profile = await SellerQuerier(conn).unverify_seller(
 473        user_id=seller_id
 474    )
 475    if not unverified_seller_profile:
 476        raise HTTPException(status.HTTP_404_NOT_FOUND, "Seller not found")
 477    return unverified_seller_profile
 478
 479
 480class UpdateSellerForm(BaseModel):
 481    """Seller profile update form."""
 482
 483    seller_name: str
 484    address_line1: str
 485    address_line2: str | None = None
 486    city: str
 487    post_code: str
 488    region: str | None = None
 489    country: str
 490    latitude: float | None = None
 491    longitude: float | None = None
 492
 493
 494@router.patch(
 495    "/database/sellers/{seller_id}",
 496    status_code=status.HTTP_200_OK,
 497    summary="Update seller profile",
 498    description="Updates the profile information for a specific seller.",
 499)
 500async def update_seller_profile(
 501    seller_id: int, form: UpdateSellerForm, conn: database_dependency, _: AdminAuthDep
 502) -> Seller:
 503    """Update seller profile.
 504
 505    Args:
 506        seller_id: seller id
 507        form: seller update form
 508        conn: database connection
 509
 510    Returns:
 511        updated seller
 512
 513    Raises:
 514        HTTPException: if seller not found or invalid coordinate update
 515    """
 516    seller_querier = SellerQuerier(conn)
 517    current_seller = await seller_querier.get_seller(user_id=seller_id)
 518    if not current_seller:
 519        raise HTTPException(status.HTTP_404_NOT_FOUND, "Seller not found")
 520
 521    if current_seller.verified_by is not None and (
 522        form.latitude is None or form.longitude is None
 523    ):
 524        raise HTTPException(
 525            status_code=status.HTTP_400_BAD_REQUEST,
 526            detail="Cannot remove coordinates from a verified seller. "
 527            "Unverify them first.",
 528        )
 529
 530    updated_seller_profile = await seller_querier.update_seller(
 531        UpdateSellerParams(
 532            user_id=seller_id,
 533            seller_name=form.seller_name,
 534            address_line1=form.address_line1,
 535            address_line2=form.address_line2,
 536            city=form.city,
 537            post_code=form.post_code,
 538            region=form.region,
 539            country=form.country,
 540            latitude=form.latitude,
 541            longitude=form.longitude,
 542        )
 543    )
 544    if not updated_seller_profile:
 545        raise HTTPException(status.HTTP_404_NOT_FOUND, "Seller not found")
 546    return updated_seller_profile
 547
 548
 549@router.get(
 550    "/database/consumers",
 551    status_code=status.HTTP_200_OK,
 552    summary="Get all consumers",
 553    description="Retrieves a list of all consumers in the system.",
 554)
 555async def get_all_consumers(
 556    conn: database_dependency, _: AdminAuthDep
 557) -> list[GetConsumersRow]:
 558    """Get all consumers.
 559
 560    Args:
 561        conn: database connection
 562
 563    Returns:
 564        list of all consumers
 565    """
 566    return [
 567        consumer_row async for consumer_row in ConsumerQuerier(conn).get_consumers()
 568    ]
 569
 570
 571class UpdateConsumerForm(BaseModel):
 572    """Consumer profile update form."""
 573
 574    first_name: str
 575    last_name: str
 576
 577
 578@router.patch(
 579    "/database/consumers/{consumer_id}",
 580    status_code=status.HTTP_200_OK,
 581    summary="Update consumer profile",
 582    description="Updates the profile information for a specific consumer.",
 583)
 584async def update_consumer_profile(
 585    consumer_id: int,
 586    form: UpdateConsumerForm,
 587    conn: database_dependency,
 588    _: AdminAuthDep,
 589) -> Consumer:
 590    """Update consumer profile.
 591
 592    Args:
 593        consumer_id: consumer id
 594        form: consumer update form
 595        conn: database connection
 596
 597    Returns:
 598        updated consumer
 599
 600    Raises:
 601        HTTPException: if consumer not found
 602    """
 603    updated_consumer_profile = await ConsumerQuerier(conn).update_consumer(
 604        UpdateConsumerParams(
 605            user_id=consumer_id, fname=form.first_name, lname=form.last_name
 606        )
 607    )
 608    if not updated_consumer_profile:
 609        raise HTTPException(status.HTTP_404_NOT_FOUND, "Consumer not found")
 610    return updated_consumer_profile
 611
 612
 613@router.get(
 614    "/database/bundles",
 615    status_code=status.HTTP_200_OK,
 616    summary="Get all bundles",
 617    description="Retrieves a list of all bundles in the system.",
 618)
 619async def get_all_bundles(conn: database_dependency, _: AdminAuthDep) -> list[Bundle]:
 620    """Get all bundles.
 621
 622    Args:
 623        conn: database connection
 624
 625    Returns:
 626        list of all bundles
 627    """
 628    return [bundle_row async for bundle_row in BundleQuerier(conn).get_bundles()]
 629
 630
 631@router.delete(
 632    "/database/bundles/{bundle_id}",
 633    status_code=status.HTTP_200_OK,
 634    summary="Delete bundle",
 635    description="Deletes a specific bundle from the system.",
 636)
 637async def delete_bundle(
 638    bundle_id: int, conn: database_dependency, _: AdminAuthDep
 639) -> Bundle:
 640    """Delete bundle.
 641
 642    Args:
 643        bundle_id: bundle id
 644        conn: database connection
 645
 646    Returns:
 647        deleted bundle
 648
 649    Raises:
 650        HTTPException: if bundle not found
 651    """
 652    deleted_bundle = await BundleQuerier(conn).delete_bundle(bundle_id=bundle_id)
 653    if not deleted_bundle:
 654        raise HTTPException(status.HTTP_404_NOT_FOUND, "Bundle not found")
 655    return deleted_bundle
 656
 657
 658@router.get(
 659    "/database/reservations",
 660    status_code=status.HTTP_200_OK,
 661    summary="Get all reservations",
 662    description="Retrieves a list of all reservations in the system.",
 663)
 664async def get_all_reservations(
 665    conn: database_dependency, _: AdminAuthDep
 666) -> list[Reservation]:
 667    """Get all reservations.
 668
 669    Args:
 670        conn: database connection
 671
 672    Returns:
 673        list of all reservations
 674    """
 675    return [
 676        reservation_row
 677        async for reservation_row in ReservationsQuerier(conn).get_reservations()
 678    ]
 679
 680
 681@router.delete(
 682    "/database/reservations/{reservation_id}",
 683    status_code=status.HTTP_200_OK,
 684    summary="Delete reservation",
 685    description="Deletes a specific reservation from the system.",
 686)
 687async def delete_reservation(
 688    reservation_id: int, conn: database_dependency, _: AdminAuthDep
 689) -> Reservation:
 690    """Delete reservation.
 691
 692    Args:
 693        reservation_id: reservation id
 694        conn: database connection
 695
 696    Returns:
 697        deleted reservation
 698
 699    Raises:
 700        HTTPException: if reservation not found
 701    """
 702    deleted_reservation = await ReservationsQuerier(conn).delete_reservation(
 703        reservation_id=reservation_id
 704    )
 705    if not deleted_reservation:
 706        raise HTTPException(status.HTTP_404_NOT_FOUND, "Reservation not found")
 707    return deleted_reservation
 708
 709
 710@router.get(
 711    "/database/allergens",
 712    status_code=status.HTTP_200_OK,
 713    summary="Get all allergens",
 714    description="Retrieves a list of all allergens in the system.",
 715)
 716async def get_all_allergens(
 717    conn: database_dependency, _: AdminAuthDep
 718) -> list[Allergen]:
 719    """Get all allergens.
 720
 721    Args:
 722        conn: database connection
 723
 724    Returns:
 725        list of all allergens
 726    """
 727    return [
 728        allergen_row async for allergen_row in AllergensQuerier(conn).get_allergens()
 729    ]
 730
 731
 732class CreateAllergenForm(BaseModel):
 733    """Allergen creation form."""
 734
 735    allergen_name: str
 736
 737
 738@router.post(
 739    "/database/allergens",
 740    status_code=status.HTTP_201_CREATED,
 741    summary="Create allergen",
 742    description="Creates a new allergen in the system.",
 743)
 744async def create_allergen(
 745    form: CreateAllergenForm, conn: database_dependency, _: AdminAuthDep
 746) -> Allergen:
 747    """Create allergen.
 748
 749    Args:
 750        form: form with the name of the allergen
 751        conn: database connection
 752
 753    Returns:
 754        created allergen
 755
 756    Raises:
 757        HTTPException: if failed to create allergen
 758    """
 759    created_allergen = await AllergensQuerier(conn).create_allergen(
 760        allergen_name=form.allergen_name
 761    )
 762    if not created_allergen:
 763        raise HTTPException(
 764            status.HTTP_500_INTERNAL_SERVER_ERROR, "Failed to create allergen"
 765        )
 766    return created_allergen
 767
 768
 769class UpdateAllergenForm(BaseModel):
 770    """Allergen update form."""
 771
 772    allergen_name: str
 773
 774
 775@router.patch(
 776    "/database/allergens/{allergen_id}",
 777    status_code=status.HTTP_200_OK,
 778    summary="Update allergen",
 779    description="Updates the name of a specific allergen.",
 780)
 781async def update_allergen(
 782    allergen_id: int,
 783    form: UpdateAllergenForm,
 784    conn: database_dependency,
 785    _: AdminAuthDep,
 786) -> Allergen:
 787    """Update allergen.
 788
 789    Args:
 790        allergen_id: allergen id
 791        form: form with the new name of the allergen
 792        conn: database connection
 793
 794    Returns:
 795        updated allergen
 796
 797    Raises:
 798        HTTPException: if allergen not found
 799    """
 800    updated_allergen = await AllergensQuerier(conn).update_allergen(
 801        UpdateAllergenParams(allergen_id=allergen_id, allergen_name=form.allergen_name)
 802    )
 803    if not updated_allergen:
 804        raise HTTPException(status.HTTP_404_NOT_FOUND, "Allergen not found")
 805    return updated_allergen
 806
 807
 808@router.delete(
 809    "/database/allergens/{allergen_id}",
 810    status_code=status.HTTP_200_OK,
 811    summary="Delete allergen",
 812    description="Deletes a specific allergen from the system.",
 813)
 814async def delete_allergen(
 815    allergen_id: int, conn: database_dependency, _: AdminAuthDep
 816) -> Allergen:
 817    """Delete allergen.
 818
 819    Args:
 820        allergen_id: allergen id
 821        conn: database connection
 822
 823    Returns:
 824        deleted allergen
 825
 826    Raises:
 827        HTTPException: if allergen not found
 828    """
 829    deleted_allergen = await AllergensQuerier(conn).delete_allergen(
 830        allergen_id=allergen_id
 831    )
 832    if not deleted_allergen:
 833        raise HTTPException(status.HTTP_404_NOT_FOUND, "Allergen not found")
 834    return deleted_allergen
 835
 836
 837@router.get(
 838    "/database/categories",
 839    status_code=status.HTTP_200_OK,
 840    summary="Get all categories",
 841    description="Retrieves a list of all categories in the system.",
 842)
 843async def get_all_categories(
 844    conn: database_dependency, _: AdminAuthDep
 845) -> list[Category]:
 846    """Get all categories.
 847
 848    Args:
 849        conn: database connection
 850
 851    Returns:
 852        list of all categories
 853    """
 854    return [
 855        category_row async for category_row in CategoryQuerier(conn).get_categories()
 856    ]
 857
 858
 859class CreateCategoryForm(BaseModel):
 860    """Category creation form."""
 861
 862    category_name: str
 863    category_coefficient: float
 864
 865
 866@router.post(
 867    "/database/categories",
 868    status_code=status.HTTP_201_CREATED,
 869    summary="Create category",
 870    description="Creates a new category in the system.",
 871)
 872async def create_category(
 873    form: CreateCategoryForm, conn: database_dependency, _: AdminAuthDep
 874) -> Category:
 875    """Create category.
 876
 877    Args:
 878        form: category creation form
 879        conn: database connection
 880
 881    Returns:
 882        created category
 883
 884    Raises:
 885        HTTPException: if failed to create category
 886    """
 887    created_category = await CategoryQuerier(conn).create_category(
 888        CreateCategoryParams(
 889            category_name=form.category_name,
 890            category_coefficient=form.category_coefficient,
 891        )
 892    )
 893    if not created_category:
 894        raise HTTPException(
 895            status.HTTP_500_INTERNAL_SERVER_ERROR, "Failed to create category"
 896        )
 897    return created_category
 898
 899
 900@router.patch(
 901    "/database/categories/{category_id}",
 902    status_code=status.HTTP_200_OK,
 903    summary="Update category",
 904    description="Updates the name and coefficient of a specific category.",
 905)
 906async def update_category(
 907    category_id: int,
 908    form: CreateCategoryForm,
 909    conn: database_dependency,
 910    _: AdminAuthDep,
 911) -> Category:
 912    """Update category.
 913
 914    Args:
 915        category_id: category id
 916        form: category update form
 917        conn: database connection
 918
 919    Returns:
 920        updated category
 921
 922    Raises:
 923        HTTPException: if category not found
 924    """
 925    updated_category = await CategoryQuerier(conn).update_category(
 926        UpdateCategoryParams(
 927            category_id=category_id,
 928            category_name=form.category_name,
 929            category_coefficient=form.category_coefficient,
 930        )
 931    )
 932    if not updated_category:
 933        raise HTTPException(status.HTTP_404_NOT_FOUND, "Category not found")
 934    return updated_category
 935
 936
 937@router.delete(
 938    "/database/categories/{category_id}",
 939    status_code=status.HTTP_200_OK,
 940    summary="Delete category",
 941    description="Deletes a specific category from the system.",
 942)
 943async def delete_category(
 944    category_id: int, conn: database_dependency, _: AdminAuthDep
 945) -> Category:
 946    """Delete category.
 947
 948    Args:
 949        category_id: category id
 950        conn: database connection
 951
 952    Returns:
 953        deleted category
 954
 955    Raises:
 956        HTTPException: if category not found
 957    """
 958    deleted_category = await CategoryQuerier(conn).delete_category(
 959        category_id=category_id
 960    )
 961    if not deleted_category:
 962        raise HTTPException(status.HTTP_404_NOT_FOUND, "Category not found")
 963    return deleted_category
 964
 965
 966@router.get(
 967    "/database/badges",
 968    status_code=status.HTTP_200_OK,
 969    summary="Get all badges",
 970    description="Retrieves a list of all badges in the system.",
 971)
 972async def get_all_badges(conn: database_dependency, _: AdminAuthDep) -> list[Badge]:
 973    """Get all badges.
 974
 975    Args:
 976        conn: database connection
 977
 978    Returns:
 979        list of all badges
 980    """
 981    return [badge_row async for badge_row in BadgeQuerier(conn).get_badges()]
 982
 983
 984class UpdateBadgeForm(BaseModel):
 985    """Badge update form."""
 986
 987    name: str
 988    description: str
 989
 990
 991@router.patch(
 992    "/database/badges/{badge_id}",
 993    status_code=status.HTTP_200_OK,
 994    summary="Update badge",
 995    description="Updates the name and description of a specific badge.",
 996)
 997async def update_badge(
 998    badge_id: int, form: UpdateBadgeForm, conn: database_dependency, _: AdminAuthDep
 999) -> Badge:
1000    """Update badge.
1001
1002    Args:
1003        badge_id: badge id
1004        form: badge update form
1005        conn: database connection
1006
1007    Returns:
1008        updated badge
1009
1010    Raises:
1011        HTTPException: if badge not found
1012    """
1013    updated_badge = await BadgeQuerier(conn).update_badge(
1014        UpdateBadgeParams(
1015            badge_id=badge_id, name=form.name, description=form.description
1016        )
1017    )
1018    if not updated_badge:
1019        raise HTTPException(status.HTTP_404_NOT_FOUND, "Badge not found")
1020    return updated_badge
1021
1022
1023@router.get(
1024    "/database/reports/admin",
1025    status_code=status.HTTP_200_OK,
1026    summary="Get all admin issue reports",
1027    description="Retrieves a list of all admin issue reports in the system.",
1028)
1029async def get_admin_reports(
1030    conn: database_dependency, _: AdminAuthDep
1031) -> list[AdminIssueReport]:
1032    """Get all admin issue reports.
1033
1034    Args:
1035        conn: database connection
1036
1037    Returns:
1038        list of all admin issue reports
1039    """
1040    return [
1041        report_row
1042        async for report_row in AdminIssueReportsQuerier(conn).get_admin_issue_reports()
1043    ]
1044
1045
1046class UpdateReportStatusForm(BaseModel):
1047    """Issue report status update form."""
1048
1049    status: IssueStatus
1050
1051
1052@router.patch(
1053    "/database/reports/admin/{report_id}/status",
1054    status_code=status.HTTP_200_OK,
1055    summary="Update admin issue report status",
1056    description="Updates the status of a specific admin issue report.",
1057)
1058async def update_admin_report_status(
1059    report_id: int,
1060    form: UpdateReportStatusForm,
1061    conn: database_dependency,
1062    _: AdminAuthDep,
1063) -> AdminIssueReport:
1064    """Update admin issue report status.
1065
1066    Args:
1067        report_id: report id
1068        form: new status form
1069        conn: database connection
1070
1071    Returns:
1072        updated report
1073
1074    Raises:
1075        HTTPException: if report not found
1076    """
1077    updated_report = await AdminIssueReportsQuerier(
1078        conn
1079    ).update_admin_issue_report_status(
1080        UpdateAdminIssueReportStatusParams(report_id=report_id, status=form.status)
1081    )
1082    if not updated_report:
1083        raise HTTPException(status.HTTP_404_NOT_FOUND, "Report not found")
1084    return updated_report
1085
1086
1087@router.get(
1088    "/database/reports/seller",
1089    status_code=status.HTTP_200_OK,
1090    summary="Get all seller issue reports",
1091    description="Retrieves a list of all seller issue reports in the system.",
1092)
1093async def get_seller_reports(
1094    conn: database_dependency, _: AdminAuthDep
1095) -> list[SellerIssueReport]:
1096    """Get all seller issue reports.
1097
1098    Args:
1099        conn: database connection
1100
1101    Returns:
1102        list of all seller issue reports
1103    """
1104    return [
1105        report_row
1106        async for report_row in SellerIssueReportsQuerier(
1107            conn
1108        ).get_seller_issue_reports()
1109    ]
1110
1111
1112@router.patch(
1113    "/database/reports/seller/{report_id}/status",
1114    status_code=status.HTTP_200_OK,
1115    summary="Update seller issue report status",
1116    description="Updates the status of a specific seller issue report.",
1117)
1118async def update_seller_report_status(
1119    report_id: int,
1120    form: UpdateReportStatusForm,
1121    conn: database_dependency,
1122    _: AdminAuthDep,
1123) -> SellerIssueReport:
1124    """Update seller issue report status.
1125
1126    Args:
1127        report_id: report id
1128        form: new status form
1129        conn: database connection
1130
1131    Returns:
1132        updated report
1133
1134    Raises:
1135        HTTPException: if report not found
1136    """
1137    updated_report = await SellerIssueReportsQuerier(
1138        conn
1139    ).update_seller_issue_report_status(
1140        UpdateSellerIssueReportStatusParams(report_id=report_id, status=form.status)
1141    )
1142    if not updated_report:
1143        raise HTTPException(status.HTTP_404_NOT_FOUND, "Report not found")
1144    return updated_report
1145
1146
1147@router.get(
1148    "/database/inbox",
1149    status_code=status.HTTP_200_OK,
1150    summary="Get all inbox messages",
1151    description="Retrieves a list of all inbox messages in the system.",
1152)
1153async def get_all_inboxes(conn: database_dependency, _: AdminAuthDep) -> list[Inbox]:
1154    """Get all inbox messages.
1155
1156    Args:
1157        conn: database connection
1158
1159    Returns:
1160        list of all inbox messages
1161    """
1162    return [inbox_row async for inbox_row in InboxQuerier(conn).get_inboxes()]
1163
1164
1165@router.get(
1166    "/database/inbox/user/{user_id}",
1167    status_code=status.HTTP_200_OK,
1168    summary="Get user inbox",
1169    description="Retrieves all inbox messages for a specific user.",
1170)
1171async def get_user_inbox(
1172    user_id: int, conn: database_dependency, _: AdminAuthDep
1173) -> list[Inbox]:
1174    """Get all inbox messages for a specific user.
1175
1176    Args:
1177        user_id: user id
1178        conn: database connection
1179
1180    Returns:
1181        list of all inbox messages for the user
1182    """
1183    return [
1184        inbox_row
1185        async for inbox_row in InboxQuerier(conn).get_user_inbox(user_id=user_id)
1186    ]
1187
1188
1189class CreateInboxMessageForm(BaseModel):
1190    """Inbox message creation form."""
1191
1192    user_id: int
1193    sender_id: int
1194    message_subject: str
1195    message_text: str
1196
1197
1198@router.post(
1199    "/database/inbox",
1200    status_code=status.HTTP_201_CREATED,
1201    summary="Create inbox message",
1202    description="Creates a new inbox message in the system.",
1203)
1204async def create_inbox_message(
1205    form: CreateInboxMessageForm, conn: database_dependency, _: AdminAuthDep
1206) -> Inbox:
1207    """Create an inbox message.
1208
1209    Args:
1210        form: form with the message details
1211        conn: database connection
1212
1213    Returns:
1214        created inbox message
1215
1216    Raises:
1217        HTTPException: if failed to create message
1218    """
1219    created_message = await InboxQuerier(conn).create_inbox_message(
1220        CreateInboxMessageParams(
1221            user_id=form.user_id,
1222            sender_id=form.sender_id,
1223            message_subject=form.message_subject,
1224            message_text=form.message_text,
1225        )
1226    )
1227    if not created_message:
1228        raise HTTPException(
1229            status.HTTP_500_INTERNAL_SERVER_ERROR, "Failed to create message"
1230        )
1231    return created_message
1232
1233
1234@router.delete(
1235    "/database/inbox/{message_id}",
1236    status_code=status.HTTP_200_OK,
1237    summary="Delete inbox message",
1238    description="Deletes a specific inbox message from the system.",
1239)
1240async def delete_inbox_message(
1241    message_id: int, conn: database_dependency, _: AdminAuthDep
1242) -> Inbox:
1243    """Delete an inbox message.
1244
1245    Args:
1246        message_id: message id
1247        conn: database connection
1248
1249    Returns:
1250        deleted message
1251
1252    Raises:
1253        HTTPException: if message not found
1254    """
1255    deleted_message = await InboxQuerier(conn).delete_inbox_message(
1256        message_id=message_id
1257    )
1258    if not deleted_message:
1259        raise HTTPException(status.HTTP_404_NOT_FOUND, "Message not found")
1260    return deleted_message
router = <fastapi.routing.APIRouter object>
class UpdateAdminForm(pydantic.main.BaseModel): 
70class UpdateAdminForm(BaseModel):
71    """Admin name update form."""
72
73    first_name: str
74    last_name: str

Admin name update form.

first_name: str = PydanticUndefined
last_name: str = PydanticUndefined
@router.post('', status_code=status.HTTP_201_CREATED, summary='Create admin', description='Create admin by root user', tags=['root admin'])
async def register_admin( form: internal.auth.creation.CreateAdminForm, 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)], _: Annotated[NoneType, Security(dependency=<function root_auth>, use_cache=True, scope=None, scopes=None)]) -> None: 
77@router.post(
78    "",
79    status_code=status.HTTP_201_CREATED,
80    summary="Create admin",
81    description="Create admin by root user",
82    tags=["root admin"],
83)
84async def register_admin(
85    form: CreateAdminForm, conn: database_dependency, _: RootAuthDep
86) -> None:
87    """Create admin by root user.
88
89    Args:
90        form: new admin information
91        conn: database connection
92    """
93    await create_admin(form, conn)

Create admin by root user.

Arguments:
  • form: new admin information
  • conn: database connection
@router.get('', status_code=status.HTTP_200_OK, summary='Get all admins', description='Retrieves a list of all registered admins by root user.', tags=['root admin'])
async def get_admins( 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)], _: Annotated[NoneType, Security(dependency=<function root_auth>, use_cache=True, scope=None, scopes=None)]) -> list[internal.queries.admin.GetAdminsRow]: 
 96@router.get(
 97    "",
 98    status_code=status.HTTP_200_OK,
 99    summary="Get all admins",
100    description="Retrieves a list of all registered admins by root user.",
101    tags=["root admin"],
102)
103async def get_admins(conn: database_dependency, _: RootAuthDep) -> list[GetAdminsRow]:
104    """Get all admins by root user.
105
106    Args:
107        conn: database connection
108
109    Returns:
110        list of all admins
111    """
112    return [admin_row async for admin_row in AdminQuerier(conn).get_admins()]

Get all admins by root user.

Arguments:
  • conn: database connection
Returns:

list of all admins

@router.get('/me', status_code=status.HTTP_200_OK, summary='Get authenticated admin', description='Retrieves the profile of the authenticated admin.')
async def get_admin_me( 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)], admin_session: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.admin.GetAdminRow: 
115@router.get(
116    "/me",
117    status_code=status.HTTP_200_OK,
118    summary="Get authenticated admin",
119    description="Retrieves the profile of the authenticated admin.",
120)
121async def get_admin_me(
122    conn: database_dependency, admin_session: AdminAuthDep
123) -> GetAdminRow:
124    """Get authenticated admin profile.
125
126    Args:
127        conn: database connection
128        admin_session: admin session
129
130    Returns:
131        admin profile
132
133    Raises:
134        HTTPException: if admin not found
135    """
136    admin_profile = await AdminQuerier(conn).get_admin(user_id=admin_session.user_id)
137    if not admin_profile:
138        raise HTTPException(
139            status_code=status.HTTP_404_NOT_FOUND, detail="Admin profile not found"
140        )
141    return admin_profile

Get authenticated admin profile.

Arguments:
  • conn: database connection
  • admin_session: admin session
Returns:

admin profile

Raises:
  • HTTPException: if admin not found
@router.get('/{admin_id}', status_code=status.HTTP_200_OK, summary='Get admin by ID', description='Retrieves the profile of an admin by their unique ID by root user.', tags=['root admin'])
async def get_admin_by_id( admin_id: int, 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)], _: Annotated[NoneType, Security(dependency=<function root_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.admin.GetAdminRow: 
144@router.get(
145    "/{admin_id}",
146    status_code=status.HTTP_200_OK,
147    summary="Get admin by ID",
148    description="Retrieves the profile of an admin by their unique ID by root user.",
149    tags=["root admin"],
150)
151async def get_admin_by_id(
152    admin_id: int, conn: database_dependency, _: RootAuthDep
153) -> GetAdminRow:
154    """Get admin profile by ID by root user.
155
156    Args:
157        admin_id: unique identifier of the admin
158        conn: database connection
159
160    Returns:
161        admin profile
162
163    Raises:
164        HTTPException: if admin not found
165    """
166    admin_profile = await AdminQuerier(conn).get_admin(user_id=admin_id)
167    if not admin_profile:
168        raise HTTPException(
169            status_code=status.HTTP_404_NOT_FOUND, detail="Admin not found"
170        )
171    return admin_profile

Get admin profile by ID by root user.

Arguments:
  • admin_id: unique identifier of the admin
  • conn: database connection
Returns:

admin profile

Raises:
  • HTTPException: if admin not found
@router.patch('/{admin_id}', status_code=status.HTTP_200_OK, summary='Update admin profile', description='Updates the profile information for an admin by root user.', tags=['root admin'])
async def update_admin( admin_id: int, form: UpdateAdminForm, 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)], _: Annotated[NoneType, Security(dependency=<function root_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.models.Admin: 
174@router.patch(
175    "/{admin_id}",
176    status_code=status.HTTP_200_OK,
177    summary="Update admin profile",
178    description="Updates the profile information for an admin by root user.",
179    tags=["root admin"],
180)
181async def update_admin(
182    admin_id: int, form: UpdateAdminForm, conn: database_dependency, _: RootAuthDep
183) -> Admin:
184    """Admin name update by root user.
185
186    Args:
187        admin_id: admin id
188        form: admin update form
189        conn: database connection
190
191    Returns:
192        updated admin
193
194    Raises:
195        HTTPException: if failed to update admin
196    """
197    updated_admin_profile = await AdminQuerier(conn).update_admin(
198        UpdateAdminParams(user_id=admin_id, fname=form.first_name, lname=form.last_name)
199    )
200    if not updated_admin_profile:
201        raise HTTPException(
202            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
203            detail="Failed to update admin",
204        )
205    return updated_admin_profile

Admin name update by root user.

Arguments:
  • admin_id: admin id
  • form: admin update form
  • conn: database connection
Returns:

updated admin

Raises:
  • HTTPException: if failed to update admin
@router.patch('/{admin_id}/deactivate', status_code=status.HTTP_200_OK, summary='Deactivate admin', description='Deactivate admin by root user', tags=['root admin'])
async def deactivate_admin( admin_id: int, 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)], _: Annotated[NoneType, Security(dependency=<function root_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.models.Admin: 
208@router.patch(
209    "/{admin_id}/deactivate",
210    status_code=status.HTTP_200_OK,
211    summary="Deactivate admin",
212    description="Deactivate admin by root user",
213    tags=["root admin"],
214)
215async def deactivate_admin(
216    admin_id: int, conn: database_dependency, _: RootAuthDep
217) -> Admin:
218    """Deactivate admin.
219
220    Args:
221        admin_id: admin id
222        conn: database connection
223
224    Returns:
225        deactivated admin
226
227    Raises:
228        HTTPException: if failed to find admin
229    """
230    admin_deactivation_result = await AdminQuerier(conn).set_is_admin_active(
231        SetIsAdminActiveParams(user_id=admin_id, active=False)
232    )
233    if not admin_deactivation_result:
234        raise HTTPException(status.HTTP_400_BAD_REQUEST, "No admin was found")
235    return admin_deactivation_result

Deactivate admin.

Arguments:
  • admin_id: admin id
  • conn: database connection
Returns:

deactivated admin

Raises:
  • HTTPException: if failed to find admin
@router.patch('/{admin_id}/activate', status_code=status.HTTP_200_OK, summary='Activate admin', description='Activate admin by root user', tags=['root admin'])
async def activate_admin( admin_id: int, 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)], _: Annotated[NoneType, Security(dependency=<function root_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.models.Admin: 
238@router.patch(
239    "/{admin_id}/activate",
240    status_code=status.HTTP_200_OK,
241    summary="Activate admin",
242    description="Activate admin by root user",
243    tags=["root admin"],
244)
245async def activate_admin(
246    admin_id: int, conn: database_dependency, _: RootAuthDep
247) -> Admin:
248    """Activate admin.
249
250    Args:
251        admin_id: admin id
252        conn: database connection
253
254    Returns:
255        activated admin
256
257    Raises:
258        HTTPException: if failed to find admin
259    """
260    admin_activation_result = await AdminQuerier(conn).set_is_admin_active(
261        SetIsAdminActiveParams(user_id=admin_id, active=True)
262    )
263    if not admin_activation_result:
264        raise HTTPException(status.HTTP_400_BAD_REQUEST, "No admin was found")
265    return admin_activation_result

Activate admin.

Arguments:
  • admin_id: admin id
  • conn: database connection
Returns:

activated admin

Raises:
  • HTTPException: if failed to find admin
@router.get('/database/users', status_code=status.HTTP_200_OK, summary='Get all users', description='Retrieves a list of all users in the system.')
async def get_all_users( 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> list[internal.queries.user.GetUsersRow]: 
268@router.get(
269    "/database/users",
270    status_code=status.HTTP_200_OK,
271    summary="Get all users",
272    description="Retrieves a list of all users in the system.",
273)
274async def get_all_users(
275    conn: database_dependency, _: AdminAuthDep
276) -> list[GetUsersRow]:
277    """Get all users.
278
279    Args:
280        conn: database connection
281
282    Returns:
283        list of all users
284    """
285    return [user_row async for user_row in UserQuerier(conn).get_users()]

Get all users.

Arguments:
  • conn: database connection
Returns:

list of all users

class UpdateUserEmailForm(pydantic.main.BaseModel): 
288class UpdateUserEmailForm(BaseModel):
289    """User email update form."""
290
291    email: str

User email update form.

email: str = PydanticUndefined
@router.patch('/database/users/{user_id}/email', status_code=status.HTTP_200_OK, summary='Update user email', description='Updates the email address for a specific user.')
async def update_user_email( user_id: int, form: UpdateUserEmailForm, 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.user.UpdateUserEmailRow: 
294@router.patch(
295    "/database/users/{user_id}/email",
296    status_code=status.HTTP_200_OK,
297    summary="Update user email",
298    description="Updates the email address for a specific user.",
299)
300async def update_user_email(
301    user_id: int, form: UpdateUserEmailForm, conn: database_dependency, _: AdminAuthDep
302) -> UpdateUserEmailRow:
303    """Update user email.
304
305    Args:
306        user_id: user id
307        form: email update form
308        conn: database connection
309
310    Returns:
311        updated user record
312
313    Raises:
314        HTTPException: if failed to update email
315    """
316    updated_user_email = await UserQuerier(conn).update_user_email(
317        UpdateUserEmailParams(user_id=user_id, email=form.email)
318    )
319    if not updated_user_email:
320        raise HTTPException(status.HTTP_404_NOT_FOUND, "User not found")
321    return updated_user_email

Update user email.

Arguments:
  • user_id: user id
  • form: email update form
  • conn: database connection
Returns:

updated user record

Raises:
  • HTTPException: if failed to update email
class UpdateUserPasswordForm(pydantic.main.BaseModel): 
324class UpdateUserPasswordForm(BaseModel):
325    """User password update form."""
326
327    password: SecretStr

User password update form.

password: pydantic.types.SecretStr = PydanticUndefined
@router.patch('/database/users/{user_id}/password', status_code=status.HTTP_200_OK, summary='Update user password', description='Updates the password for a specific user.')
async def update_user_password( user_id: int, form: UpdateUserPasswordForm, 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.user.UpdateUserPasswordRow: 
330@router.patch(
331    "/database/users/{user_id}/password",
332    status_code=status.HTTP_200_OK,
333    summary="Update user password",
334    description="Updates the password for a specific user.",
335)
336async def update_user_password(
337    user_id: int,
338    form: UpdateUserPasswordForm,
339    conn: database_dependency,
340    _: AdminAuthDep,
341) -> UpdateUserPasswordRow:
342    """Update user password.
343
344    Args:
345        user_id: user id
346        form: password update form
347        conn: database connection
348
349    Returns:
350        updated user record
351
352    Raises:
353        HTTPException: if failed to update password
354    """
355    hashed_pw = hash_password(form.password.get_secret_value())
356    updated_user_password = await UserQuerier(conn).update_user_password(
357        UpdateUserPasswordParams(user_id=user_id, pw_hash=hashed_pw)
358    )
359    if not updated_user_password:
360        raise HTTPException(status.HTTP_404_NOT_FOUND, "User not found")
361    return updated_user_password

Update user password.

Arguments:
  • user_id: user id
  • form: password update form
  • conn: database connection
Returns:

updated user record

Raises:
  • HTTPException: if failed to update password
@router.delete('/database/users/{user_id}', status_code=status.HTTP_200_OK, summary='Delete user', description='Deletes a specific user from the system.')
async def delete_user( user_id: int, 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.user.DeleteUserRow: 
364@router.delete(
365    "/database/users/{user_id}",
366    status_code=status.HTTP_200_OK,
367    summary="Delete user",
368    description="Deletes a specific user from the system.",
369)
370async def delete_user(
371    user_id: int, conn: database_dependency, _: AdminAuthDep
372) -> DeleteUserRow:
373    """Delete user.
374
375    Args:
376        user_id: user id
377        conn: database connection
378
379    Returns:
380        deleted user
381
382    Raises:
383        HTTPException: if user not found
384    """
385    deleted_user = await UserQuerier(conn).delete_user(user_id=user_id)
386    if not deleted_user:
387        raise HTTPException(status.HTTP_404_NOT_FOUND, "User not found")
388    return deleted_user

Delete user.

Arguments:
  • user_id: user id
  • conn: database connection
Returns:

deleted user

Raises:
  • HTTPException: if user not found
@router.get('/database/sellers', status_code=status.HTTP_200_OK, summary='Get all sellers', description='Retrieves a list of all sellers in the system.')
async def get_all_sellers( 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> list[internal.queries.seller.GetSellersRow]: 
391@router.get(
392    "/database/sellers",
393    status_code=status.HTTP_200_OK,
394    summary="Get all sellers",
395    description="Retrieves a list of all sellers in the system.",
396)
397async def get_all_sellers(
398    conn: database_dependency, _: AdminAuthDep
399) -> list[GetSellersRow]:
400    """Get all sellers.
401
402    Args:
403        conn: database connection
404
405    Returns:
406        list of all sellers
407    """
408    return [seller_row async for seller_row in SellerQuerier(conn).get_sellers()]

Get all sellers.

Arguments:
  • conn: database connection
Returns:

list of all sellers

@router.patch('/database/sellers/{seller_id}/verify', status_code=status.HTTP_200_OK, summary='Verify seller', description='Verifies a seller. Only possible if coordinates exist.')
async def verify_seller( seller_id: int, 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)], admin_session: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.models.Seller: 
411@router.patch(
412    "/database/sellers/{seller_id}/verify",
413    status_code=status.HTTP_200_OK,
414    summary="Verify seller",
415    description="Verifies a seller. Only possible if coordinates exist.",
416)
417async def verify_seller(
418    seller_id: int, conn: database_dependency, admin_session: AdminAuthDep
419) -> Seller:
420    """Verify seller. Only possible if coordinates exist.
421
422    Args:
423        seller_id: seller id
424        conn: database connection
425        admin_session: admin session
426
427    Returns:
428        verified seller
429
430    Raises:
431        HTTPException: if seller not found or coordinates missing
432    """
433    seller_querier = SellerQuerier(conn)
434    seller_profile = await seller_querier.get_seller(user_id=seller_id)
435    if not seller_profile:
436        raise HTTPException(status.HTTP_404_NOT_FOUND, "Seller not found")
437
438    if seller_profile.latitude is None or seller_profile.longitude is None:
439        raise HTTPException(
440            status_code=status.HTTP_400_BAD_REQUEST,
441            detail="Cannot verify seller without valid coordinates",
442        )
443
444    verified_seller_profile = await seller_querier.verify_seller(
445        VerifySellerParams(user_id=seller_id, verified_by=admin_session.user_id)
446    )
447    if not verified_seller_profile:
448        raise HTTPException(status.HTTP_404_NOT_FOUND, "Seller not found")
449    return verified_seller_profile

Verify seller. Only possible if coordinates exist.

Arguments:
  • seller_id: seller id
  • conn: database connection
  • admin_session: admin session
Returns:

verified seller

Raises:
  • HTTPException: if seller not found or coordinates missing
@router.patch('/database/sellers/{seller_id}/unverify', status_code=status.HTTP_200_OK, summary='Unverify seller', description='Removes verification status from a seller.')
async def unverify_seller( seller_id: int, 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.models.Seller: 
452@router.patch(
453    "/database/sellers/{seller_id}/unverify",
454    status_code=status.HTTP_200_OK,
455    summary="Unverify seller",
456    description="Removes verification status from a seller.",
457)
458async def unverify_seller(
459    seller_id: int, conn: database_dependency, _: AdminAuthDep
460) -> Seller:
461    """Unverify seller.
462
463    Args:
464        seller_id: seller id
465        conn: database connection
466
467    Returns:
468        unverified seller
469
470    Raises:
471        HTTPException: if seller not found
472    """
473    unverified_seller_profile = await SellerQuerier(conn).unverify_seller(
474        user_id=seller_id
475    )
476    if not unverified_seller_profile:
477        raise HTTPException(status.HTTP_404_NOT_FOUND, "Seller not found")
478    return unverified_seller_profile

Unverify seller.

Arguments:
  • seller_id: seller id
  • conn: database connection
Returns:

unverified seller

Raises:
  • HTTPException: if seller not found
class UpdateSellerForm(pydantic.main.BaseModel): 
481class UpdateSellerForm(BaseModel):
482    """Seller profile update form."""
483
484    seller_name: str
485    address_line1: str
486    address_line2: str | None = None
487    city: str
488    post_code: str
489    region: str | None = None
490    country: str
491    latitude: float | None = None
492    longitude: float | None = None

Seller profile update form.

seller_name: str = PydanticUndefined
address_line1: str = PydanticUndefined
address_line2: str | None = None
city: str = PydanticUndefined
post_code: str = PydanticUndefined
region: str | None = None
country: str = PydanticUndefined
latitude: float | None = None
longitude: float | None = None
@router.patch('/database/sellers/{seller_id}', status_code=status.HTTP_200_OK, summary='Update seller profile', description='Updates the profile information for a specific seller.')
async def update_seller_profile( seller_id: int, form: UpdateSellerForm, 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.models.Seller: 
495@router.patch(
496    "/database/sellers/{seller_id}",
497    status_code=status.HTTP_200_OK,
498    summary="Update seller profile",
499    description="Updates the profile information for a specific seller.",
500)
501async def update_seller_profile(
502    seller_id: int, form: UpdateSellerForm, conn: database_dependency, _: AdminAuthDep
503) -> Seller:
504    """Update seller profile.
505
506    Args:
507        seller_id: seller id
508        form: seller update form
509        conn: database connection
510
511    Returns:
512        updated seller
513
514    Raises:
515        HTTPException: if seller not found or invalid coordinate update
516    """
517    seller_querier = SellerQuerier(conn)
518    current_seller = await seller_querier.get_seller(user_id=seller_id)
519    if not current_seller:
520        raise HTTPException(status.HTTP_404_NOT_FOUND, "Seller not found")
521
522    if current_seller.verified_by is not None and (
523        form.latitude is None or form.longitude is None
524    ):
525        raise HTTPException(
526            status_code=status.HTTP_400_BAD_REQUEST,
527            detail="Cannot remove coordinates from a verified seller. "
528            "Unverify them first.",
529        )
530
531    updated_seller_profile = await seller_querier.update_seller(
532        UpdateSellerParams(
533            user_id=seller_id,
534            seller_name=form.seller_name,
535            address_line1=form.address_line1,
536            address_line2=form.address_line2,
537            city=form.city,
538            post_code=form.post_code,
539            region=form.region,
540            country=form.country,
541            latitude=form.latitude,
542            longitude=form.longitude,
543        )
544    )
545    if not updated_seller_profile:
546        raise HTTPException(status.HTTP_404_NOT_FOUND, "Seller not found")
547    return updated_seller_profile

Update seller profile.

Arguments:
  • seller_id: seller id
  • form: seller update form
  • conn: database connection
Returns:

updated seller

Raises:
  • HTTPException: if seller not found or invalid coordinate update
@router.get('/database/consumers', status_code=status.HTTP_200_OK, summary='Get all consumers', description='Retrieves a list of all consumers in the system.')
async def get_all_consumers( 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> list[internal.queries.consumer.GetConsumersRow]: 
550@router.get(
551    "/database/consumers",
552    status_code=status.HTTP_200_OK,
553    summary="Get all consumers",
554    description="Retrieves a list of all consumers in the system.",
555)
556async def get_all_consumers(
557    conn: database_dependency, _: AdminAuthDep
558) -> list[GetConsumersRow]:
559    """Get all consumers.
560
561    Args:
562        conn: database connection
563
564    Returns:
565        list of all consumers
566    """
567    return [
568        consumer_row async for consumer_row in ConsumerQuerier(conn).get_consumers()
569    ]

Get all consumers.

Arguments:
  • conn: database connection
Returns:

list of all consumers

class UpdateConsumerForm(pydantic.main.BaseModel): 
572class UpdateConsumerForm(BaseModel):
573    """Consumer profile update form."""
574
575    first_name: str
576    last_name: str

Consumer profile update form.

first_name: str = PydanticUndefined
last_name: str = PydanticUndefined
@router.patch('/database/consumers/{consumer_id}', status_code=status.HTTP_200_OK, summary='Update consumer profile', description='Updates the profile information for a specific consumer.')
async def update_consumer_profile( consumer_id: int, form: UpdateConsumerForm, 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.models.Consumer: 
579@router.patch(
580    "/database/consumers/{consumer_id}",
581    status_code=status.HTTP_200_OK,
582    summary="Update consumer profile",
583    description="Updates the profile information for a specific consumer.",
584)
585async def update_consumer_profile(
586    consumer_id: int,
587    form: UpdateConsumerForm,
588    conn: database_dependency,
589    _: AdminAuthDep,
590) -> Consumer:
591    """Update consumer profile.
592
593    Args:
594        consumer_id: consumer id
595        form: consumer update form
596        conn: database connection
597
598    Returns:
599        updated consumer
600
601    Raises:
602        HTTPException: if consumer not found
603    """
604    updated_consumer_profile = await ConsumerQuerier(conn).update_consumer(
605        UpdateConsumerParams(
606            user_id=consumer_id, fname=form.first_name, lname=form.last_name
607        )
608    )
609    if not updated_consumer_profile:
610        raise HTTPException(status.HTTP_404_NOT_FOUND, "Consumer not found")
611    return updated_consumer_profile

Update consumer profile.

Arguments:
  • consumer_id: consumer id
  • form: consumer update form
  • conn: database connection
Returns:

updated consumer

Raises:
  • HTTPException: if consumer not found
@router.get('/database/bundles', status_code=status.HTTP_200_OK, summary='Get all bundles', description='Retrieves a list of all bundles in the system.')
async def get_all_bundles( 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> list[internal.queries.models.Bundle]: 
614@router.get(
615    "/database/bundles",
616    status_code=status.HTTP_200_OK,
617    summary="Get all bundles",
618    description="Retrieves a list of all bundles in the system.",
619)
620async def get_all_bundles(conn: database_dependency, _: AdminAuthDep) -> list[Bundle]:
621    """Get all bundles.
622
623    Args:
624        conn: database connection
625
626    Returns:
627        list of all bundles
628    """
629    return [bundle_row async for bundle_row in BundleQuerier(conn).get_bundles()]

Get all bundles.

Arguments:
  • conn: database connection
Returns:

list of all bundles

@router.delete('/database/bundles/{bundle_id}', status_code=status.HTTP_200_OK, summary='Delete bundle', description='Deletes a specific bundle from the system.')
async def delete_bundle( bundle_id: int, 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.models.Bundle: 
632@router.delete(
633    "/database/bundles/{bundle_id}",
634    status_code=status.HTTP_200_OK,
635    summary="Delete bundle",
636    description="Deletes a specific bundle from the system.",
637)
638async def delete_bundle(
639    bundle_id: int, conn: database_dependency, _: AdminAuthDep
640) -> Bundle:
641    """Delete bundle.
642
643    Args:
644        bundle_id: bundle id
645        conn: database connection
646
647    Returns:
648        deleted bundle
649
650    Raises:
651        HTTPException: if bundle not found
652    """
653    deleted_bundle = await BundleQuerier(conn).delete_bundle(bundle_id=bundle_id)
654    if not deleted_bundle:
655        raise HTTPException(status.HTTP_404_NOT_FOUND, "Bundle not found")
656    return deleted_bundle

Delete bundle.

Arguments:
  • bundle_id: bundle id
  • conn: database connection
Returns:

deleted bundle

Raises:
  • HTTPException: if bundle not found
@router.get('/database/reservations', status_code=status.HTTP_200_OK, summary='Get all reservations', description='Retrieves a list of all reservations in the system.')
async def get_all_reservations( 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> list[internal.queries.models.Reservation]: 
659@router.get(
660    "/database/reservations",
661    status_code=status.HTTP_200_OK,
662    summary="Get all reservations",
663    description="Retrieves a list of all reservations in the system.",
664)
665async def get_all_reservations(
666    conn: database_dependency, _: AdminAuthDep
667) -> list[Reservation]:
668    """Get all reservations.
669
670    Args:
671        conn: database connection
672
673    Returns:
674        list of all reservations
675    """
676    return [
677        reservation_row
678        async for reservation_row in ReservationsQuerier(conn).get_reservations()
679    ]

Get all reservations.

Arguments:
  • conn: database connection
Returns:

list of all reservations

@router.delete('/database/reservations/{reservation_id}', status_code=status.HTTP_200_OK, summary='Delete reservation', description='Deletes a specific reservation from the system.')
async def delete_reservation( reservation_id: int, 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.models.Reservation: 
682@router.delete(
683    "/database/reservations/{reservation_id}",
684    status_code=status.HTTP_200_OK,
685    summary="Delete reservation",
686    description="Deletes a specific reservation from the system.",
687)
688async def delete_reservation(
689    reservation_id: int, conn: database_dependency, _: AdminAuthDep
690) -> Reservation:
691    """Delete reservation.
692
693    Args:
694        reservation_id: reservation id
695        conn: database connection
696
697    Returns:
698        deleted reservation
699
700    Raises:
701        HTTPException: if reservation not found
702    """
703    deleted_reservation = await ReservationsQuerier(conn).delete_reservation(
704        reservation_id=reservation_id
705    )
706    if not deleted_reservation:
707        raise HTTPException(status.HTTP_404_NOT_FOUND, "Reservation not found")
708    return deleted_reservation

Delete reservation.

Arguments:
  • reservation_id: reservation id
  • conn: database connection
Returns:

deleted reservation

Raises:
  • HTTPException: if reservation not found
@router.get('/database/allergens', status_code=status.HTTP_200_OK, summary='Get all allergens', description='Retrieves a list of all allergens in the system.')
async def get_all_allergens( 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> list[internal.queries.models.Allergen]: 
711@router.get(
712    "/database/allergens",
713    status_code=status.HTTP_200_OK,
714    summary="Get all allergens",
715    description="Retrieves a list of all allergens in the system.",
716)
717async def get_all_allergens(
718    conn: database_dependency, _: AdminAuthDep
719) -> list[Allergen]:
720    """Get all allergens.
721
722    Args:
723        conn: database connection
724
725    Returns:
726        list of all allergens
727    """
728    return [
729        allergen_row async for allergen_row in AllergensQuerier(conn).get_allergens()
730    ]

Get all allergens.

Arguments:
  • conn: database connection
Returns:

list of all allergens

class CreateAllergenForm(pydantic.main.BaseModel): 
733class CreateAllergenForm(BaseModel):
734    """Allergen creation form."""
735
736    allergen_name: str

Allergen creation form.

allergen_name: str = PydanticUndefined
@router.post('/database/allergens', status_code=status.HTTP_201_CREATED, summary='Create allergen', description='Creates a new allergen in the system.')
async def create_allergen( form: CreateAllergenForm, 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.models.Allergen: 
739@router.post(
740    "/database/allergens",
741    status_code=status.HTTP_201_CREATED,
742    summary="Create allergen",
743    description="Creates a new allergen in the system.",
744)
745async def create_allergen(
746    form: CreateAllergenForm, conn: database_dependency, _: AdminAuthDep
747) -> Allergen:
748    """Create allergen.
749
750    Args:
751        form: form with the name of the allergen
752        conn: database connection
753
754    Returns:
755        created allergen
756
757    Raises:
758        HTTPException: if failed to create allergen
759    """
760    created_allergen = await AllergensQuerier(conn).create_allergen(
761        allergen_name=form.allergen_name
762    )
763    if not created_allergen:
764        raise HTTPException(
765            status.HTTP_500_INTERNAL_SERVER_ERROR, "Failed to create allergen"
766        )
767    return created_allergen

Create allergen.

Arguments:
  • form: form with the name of the allergen
  • conn: database connection
Returns:

created allergen

Raises:
  • HTTPException: if failed to create allergen
class UpdateAllergenForm(pydantic.main.BaseModel): 
770class UpdateAllergenForm(BaseModel):
771    """Allergen update form."""
772
773    allergen_name: str

Allergen update form.

allergen_name: str = PydanticUndefined
@router.patch('/database/allergens/{allergen_id}', status_code=status.HTTP_200_OK, summary='Update allergen', description='Updates the name of a specific allergen.')
async def update_allergen( allergen_id: int, form: UpdateAllergenForm, 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.models.Allergen: 
776@router.patch(
777    "/database/allergens/{allergen_id}",
778    status_code=status.HTTP_200_OK,
779    summary="Update allergen",
780    description="Updates the name of a specific allergen.",
781)
782async def update_allergen(
783    allergen_id: int,
784    form: UpdateAllergenForm,
785    conn: database_dependency,
786    _: AdminAuthDep,
787) -> Allergen:
788    """Update allergen.
789
790    Args:
791        allergen_id: allergen id
792        form: form with the new name of the allergen
793        conn: database connection
794
795    Returns:
796        updated allergen
797
798    Raises:
799        HTTPException: if allergen not found
800    """
801    updated_allergen = await AllergensQuerier(conn).update_allergen(
802        UpdateAllergenParams(allergen_id=allergen_id, allergen_name=form.allergen_name)
803    )
804    if not updated_allergen:
805        raise HTTPException(status.HTTP_404_NOT_FOUND, "Allergen not found")
806    return updated_allergen

Update allergen.

Arguments:
  • allergen_id: allergen id
  • form: form with the new name of the allergen
  • conn: database connection
Returns:

updated allergen

Raises:
  • HTTPException: if allergen not found
@router.delete('/database/allergens/{allergen_id}', status_code=status.HTTP_200_OK, summary='Delete allergen', description='Deletes a specific allergen from the system.')
async def delete_allergen( allergen_id: int, 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.models.Allergen: 
809@router.delete(
810    "/database/allergens/{allergen_id}",
811    status_code=status.HTTP_200_OK,
812    summary="Delete allergen",
813    description="Deletes a specific allergen from the system.",
814)
815async def delete_allergen(
816    allergen_id: int, conn: database_dependency, _: AdminAuthDep
817) -> Allergen:
818    """Delete allergen.
819
820    Args:
821        allergen_id: allergen id
822        conn: database connection
823
824    Returns:
825        deleted allergen
826
827    Raises:
828        HTTPException: if allergen not found
829    """
830    deleted_allergen = await AllergensQuerier(conn).delete_allergen(
831        allergen_id=allergen_id
832    )
833    if not deleted_allergen:
834        raise HTTPException(status.HTTP_404_NOT_FOUND, "Allergen not found")
835    return deleted_allergen

Delete allergen.

Arguments:
  • allergen_id: allergen id
  • conn: database connection
Returns:

deleted allergen

Raises:
  • HTTPException: if allergen not found
@router.get('/database/categories', status_code=status.HTTP_200_OK, summary='Get all categories', description='Retrieves a list of all categories in the system.')
async def get_all_categories( 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> list[internal.queries.models.Category]: 
838@router.get(
839    "/database/categories",
840    status_code=status.HTTP_200_OK,
841    summary="Get all categories",
842    description="Retrieves a list of all categories in the system.",
843)
844async def get_all_categories(
845    conn: database_dependency, _: AdminAuthDep
846) -> list[Category]:
847    """Get all categories.
848
849    Args:
850        conn: database connection
851
852    Returns:
853        list of all categories
854    """
855    return [
856        category_row async for category_row in CategoryQuerier(conn).get_categories()
857    ]

Get all categories.

Arguments:
  • conn: database connection
Returns:

list of all categories

class CreateCategoryForm(pydantic.main.BaseModel): 
860class CreateCategoryForm(BaseModel):
861    """Category creation form."""
862
863    category_name: str
864    category_coefficient: float

Category creation form.

category_name: str = PydanticUndefined
category_coefficient: float = PydanticUndefined
@router.post('/database/categories', status_code=status.HTTP_201_CREATED, summary='Create category', description='Creates a new category in the system.')
async def create_category( form: CreateCategoryForm, 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.models.Category: 
867@router.post(
868    "/database/categories",
869    status_code=status.HTTP_201_CREATED,
870    summary="Create category",
871    description="Creates a new category in the system.",
872)
873async def create_category(
874    form: CreateCategoryForm, conn: database_dependency, _: AdminAuthDep
875) -> Category:
876    """Create category.
877
878    Args:
879        form: category creation form
880        conn: database connection
881
882    Returns:
883        created category
884
885    Raises:
886        HTTPException: if failed to create category
887    """
888    created_category = await CategoryQuerier(conn).create_category(
889        CreateCategoryParams(
890            category_name=form.category_name,
891            category_coefficient=form.category_coefficient,
892        )
893    )
894    if not created_category:
895        raise HTTPException(
896            status.HTTP_500_INTERNAL_SERVER_ERROR, "Failed to create category"
897        )
898    return created_category

Create category.

Arguments:
  • form: category creation form
  • conn: database connection
Returns:

created category

Raises:
  • HTTPException: if failed to create category
@router.patch('/database/categories/{category_id}', status_code=status.HTTP_200_OK, summary='Update category', description='Updates the name and coefficient of a specific category.')
async def update_category( category_id: int, form: CreateCategoryForm, 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.models.Category: 
901@router.patch(
902    "/database/categories/{category_id}",
903    status_code=status.HTTP_200_OK,
904    summary="Update category",
905    description="Updates the name and coefficient of a specific category.",
906)
907async def update_category(
908    category_id: int,
909    form: CreateCategoryForm,
910    conn: database_dependency,
911    _: AdminAuthDep,
912) -> Category:
913    """Update category.
914
915    Args:
916        category_id: category id
917        form: category update form
918        conn: database connection
919
920    Returns:
921        updated category
922
923    Raises:
924        HTTPException: if category not found
925    """
926    updated_category = await CategoryQuerier(conn).update_category(
927        UpdateCategoryParams(
928            category_id=category_id,
929            category_name=form.category_name,
930            category_coefficient=form.category_coefficient,
931        )
932    )
933    if not updated_category:
934        raise HTTPException(status.HTTP_404_NOT_FOUND, "Category not found")
935    return updated_category

Update category.

Arguments:
  • category_id: category id
  • form: category update form
  • conn: database connection
Returns:

updated category

Raises:
  • HTTPException: if category not found
@router.delete('/database/categories/{category_id}', status_code=status.HTTP_200_OK, summary='Delete category', description='Deletes a specific category from the system.')
async def delete_category( category_id: int, 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.models.Category: 
938@router.delete(
939    "/database/categories/{category_id}",
940    status_code=status.HTTP_200_OK,
941    summary="Delete category",
942    description="Deletes a specific category from the system.",
943)
944async def delete_category(
945    category_id: int, conn: database_dependency, _: AdminAuthDep
946) -> Category:
947    """Delete category.
948
949    Args:
950        category_id: category id
951        conn: database connection
952
953    Returns:
954        deleted category
955
956    Raises:
957        HTTPException: if category not found
958    """
959    deleted_category = await CategoryQuerier(conn).delete_category(
960        category_id=category_id
961    )
962    if not deleted_category:
963        raise HTTPException(status.HTTP_404_NOT_FOUND, "Category not found")
964    return deleted_category

Delete category.

Arguments:
  • category_id: category id
  • conn: database connection
Returns:

deleted category

Raises:
  • HTTPException: if category not found
@router.get('/database/badges', status_code=status.HTTP_200_OK, summary='Get all badges', description='Retrieves a list of all badges in the system.')
async def get_all_badges( 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> list[internal.queries.models.Badge]: 
967@router.get(
968    "/database/badges",
969    status_code=status.HTTP_200_OK,
970    summary="Get all badges",
971    description="Retrieves a list of all badges in the system.",
972)
973async def get_all_badges(conn: database_dependency, _: AdminAuthDep) -> list[Badge]:
974    """Get all badges.
975
976    Args:
977        conn: database connection
978
979    Returns:
980        list of all badges
981    """
982    return [badge_row async for badge_row in BadgeQuerier(conn).get_badges()]

Get all badges.

Arguments:
  • conn: database connection
Returns:

list of all badges

class UpdateBadgeForm(pydantic.main.BaseModel): 
985class UpdateBadgeForm(BaseModel):
986    """Badge update form."""
987
988    name: str
989    description: str

Badge update form.

name: str = PydanticUndefined
description: str = PydanticUndefined
@router.patch('/database/badges/{badge_id}', status_code=status.HTTP_200_OK, summary='Update badge', description='Updates the name and description of a specific badge.')
async def update_badge( badge_id: int, form: UpdateBadgeForm, 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.models.Badge: 
 992@router.patch(
 993    "/database/badges/{badge_id}",
 994    status_code=status.HTTP_200_OK,
 995    summary="Update badge",
 996    description="Updates the name and description of a specific badge.",
 997)
 998async def update_badge(
 999    badge_id: int, form: UpdateBadgeForm, conn: database_dependency, _: AdminAuthDep
1000) -> Badge:
1001    """Update badge.
1002
1003    Args:
1004        badge_id: badge id
1005        form: badge update form
1006        conn: database connection
1007
1008    Returns:
1009        updated badge
1010
1011    Raises:
1012        HTTPException: if badge not found
1013    """
1014    updated_badge = await BadgeQuerier(conn).update_badge(
1015        UpdateBadgeParams(
1016            badge_id=badge_id, name=form.name, description=form.description
1017        )
1018    )
1019    if not updated_badge:
1020        raise HTTPException(status.HTTP_404_NOT_FOUND, "Badge not found")
1021    return updated_badge

Update badge.

Arguments:
  • badge_id: badge id
  • form: badge update form
  • conn: database connection
Returns:

updated badge

Raises:
  • HTTPException: if badge not found
@router.get('/database/reports/admin', status_code=status.HTTP_200_OK, summary='Get all admin issue reports', description='Retrieves a list of all admin issue reports in the system.')
async def get_admin_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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> list[internal.queries.models.AdminIssueReport]: 
1024@router.get(
1025    "/database/reports/admin",
1026    status_code=status.HTTP_200_OK,
1027    summary="Get all admin issue reports",
1028    description="Retrieves a list of all admin issue reports in the system.",
1029)
1030async def get_admin_reports(
1031    conn: database_dependency, _: AdminAuthDep
1032) -> list[AdminIssueReport]:
1033    """Get all admin issue reports.
1034
1035    Args:
1036        conn: database connection
1037
1038    Returns:
1039        list of all admin issue reports
1040    """
1041    return [
1042        report_row
1043        async for report_row in AdminIssueReportsQuerier(conn).get_admin_issue_reports()
1044    ]

Get all admin issue reports.

Arguments:
  • conn: database connection
Returns:

list of all admin issue reports

class UpdateReportStatusForm(pydantic.main.BaseModel): 
1047class UpdateReportStatusForm(BaseModel):
1048    """Issue report status update form."""
1049
1050    status: IssueStatus

Issue report status update form.

status: internal.queries.models.IssueStatus = PydanticUndefined
@router.patch('/database/reports/admin/{report_id}/status', status_code=status.HTTP_200_OK, summary='Update admin issue report status', description='Updates the status of a specific admin issue report.')
async def update_admin_report_status( report_id: int, form: UpdateReportStatusForm, 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.models.AdminIssueReport: 
1053@router.patch(
1054    "/database/reports/admin/{report_id}/status",
1055    status_code=status.HTTP_200_OK,
1056    summary="Update admin issue report status",
1057    description="Updates the status of a specific admin issue report.",
1058)
1059async def update_admin_report_status(
1060    report_id: int,
1061    form: UpdateReportStatusForm,
1062    conn: database_dependency,
1063    _: AdminAuthDep,
1064) -> AdminIssueReport:
1065    """Update admin issue report status.
1066
1067    Args:
1068        report_id: report id
1069        form: new status form
1070        conn: database connection
1071
1072    Returns:
1073        updated report
1074
1075    Raises:
1076        HTTPException: if report not found
1077    """
1078    updated_report = await AdminIssueReportsQuerier(
1079        conn
1080    ).update_admin_issue_report_status(
1081        UpdateAdminIssueReportStatusParams(report_id=report_id, status=form.status)
1082    )
1083    if not updated_report:
1084        raise HTTPException(status.HTTP_404_NOT_FOUND, "Report not found")
1085    return updated_report

Update admin issue report status.

Arguments:
  • report_id: report id
  • form: new status form
  • conn: database connection
Returns:

updated report

Raises:
  • HTTPException: if report not found
@router.get('/database/reports/seller', status_code=status.HTTP_200_OK, summary='Get all seller issue reports', description='Retrieves a list of all seller issue reports in the system.')
async def get_seller_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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> list[internal.queries.models.SellerIssueReport]: 
1088@router.get(
1089    "/database/reports/seller",
1090    status_code=status.HTTP_200_OK,
1091    summary="Get all seller issue reports",
1092    description="Retrieves a list of all seller issue reports in the system.",
1093)
1094async def get_seller_reports(
1095    conn: database_dependency, _: AdminAuthDep
1096) -> list[SellerIssueReport]:
1097    """Get all seller issue reports.
1098
1099    Args:
1100        conn: database connection
1101
1102    Returns:
1103        list of all seller issue reports
1104    """
1105    return [
1106        report_row
1107        async for report_row in SellerIssueReportsQuerier(
1108            conn
1109        ).get_seller_issue_reports()
1110    ]

Get all seller issue reports.

Arguments:
  • conn: database connection
Returns:

list of all seller issue reports

@router.patch('/database/reports/seller/{report_id}/status', status_code=status.HTTP_200_OK, summary='Update seller issue report status', description='Updates the status of a specific seller issue report.')
async def update_seller_report_status( report_id: int, form: UpdateReportStatusForm, 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.models.SellerIssueReport: 
1113@router.patch(
1114    "/database/reports/seller/{report_id}/status",
1115    status_code=status.HTTP_200_OK,
1116    summary="Update seller issue report status",
1117    description="Updates the status of a specific seller issue report.",
1118)
1119async def update_seller_report_status(
1120    report_id: int,
1121    form: UpdateReportStatusForm,
1122    conn: database_dependency,
1123    _: AdminAuthDep,
1124) -> SellerIssueReport:
1125    """Update seller issue report status.
1126
1127    Args:
1128        report_id: report id
1129        form: new status form
1130        conn: database connection
1131
1132    Returns:
1133        updated report
1134
1135    Raises:
1136        HTTPException: if report not found
1137    """
1138    updated_report = await SellerIssueReportsQuerier(
1139        conn
1140    ).update_seller_issue_report_status(
1141        UpdateSellerIssueReportStatusParams(report_id=report_id, status=form.status)
1142    )
1143    if not updated_report:
1144        raise HTTPException(status.HTTP_404_NOT_FOUND, "Report not found")
1145    return updated_report

Update seller issue report status.

Arguments:
  • report_id: report id
  • form: new status form
  • conn: database connection
Returns:

updated report

Raises:
  • HTTPException: if report not found
@router.get('/database/inbox', status_code=status.HTTP_200_OK, summary='Get all inbox messages', description='Retrieves a list of all inbox messages in the system.')
async def get_all_inboxes( 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> list[internal.queries.models.Inbox]: 
1148@router.get(
1149    "/database/inbox",
1150    status_code=status.HTTP_200_OK,
1151    summary="Get all inbox messages",
1152    description="Retrieves a list of all inbox messages in the system.",
1153)
1154async def get_all_inboxes(conn: database_dependency, _: AdminAuthDep) -> list[Inbox]:
1155    """Get all inbox messages.
1156
1157    Args:
1158        conn: database connection
1159
1160    Returns:
1161        list of all inbox messages
1162    """
1163    return [inbox_row async for inbox_row in InboxQuerier(conn).get_inboxes()]

Get all inbox messages.

Arguments:
  • conn: database connection
Returns:

list of all inbox messages

@router.get('/database/inbox/user/{user_id}', status_code=status.HTTP_200_OK, summary='Get user inbox', description='Retrieves all inbox messages for a specific user.')
async def get_user_inbox( user_id: int, 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> list[internal.queries.models.Inbox]: 
1166@router.get(
1167    "/database/inbox/user/{user_id}",
1168    status_code=status.HTTP_200_OK,
1169    summary="Get user inbox",
1170    description="Retrieves all inbox messages for a specific user.",
1171)
1172async def get_user_inbox(
1173    user_id: int, conn: database_dependency, _: AdminAuthDep
1174) -> list[Inbox]:
1175    """Get all inbox messages for a specific user.
1176
1177    Args:
1178        user_id: user id
1179        conn: database connection
1180
1181    Returns:
1182        list of all inbox messages for the user
1183    """
1184    return [
1185        inbox_row
1186        async for inbox_row in InboxQuerier(conn).get_user_inbox(user_id=user_id)
1187    ]

Get all inbox messages for a specific user.

Arguments:
  • user_id: user id
  • conn: database connection
Returns:

list of all inbox messages for the user

class CreateInboxMessageForm(pydantic.main.BaseModel): 
1190class CreateInboxMessageForm(BaseModel):
1191    """Inbox message creation form."""
1192
1193    user_id: int
1194    sender_id: int
1195    message_subject: str
1196    message_text: str

Inbox message creation form.

user_id: int = PydanticUndefined
sender_id: int = PydanticUndefined
message_subject: str = PydanticUndefined
message_text: str = PydanticUndefined
@router.post('/database/inbox', status_code=status.HTTP_201_CREATED, summary='Create inbox message', description='Creates a new inbox message in the system.')
async def create_inbox_message( form: CreateInboxMessageForm, 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.models.Inbox: 
1199@router.post(
1200    "/database/inbox",
1201    status_code=status.HTTP_201_CREATED,
1202    summary="Create inbox message",
1203    description="Creates a new inbox message in the system.",
1204)
1205async def create_inbox_message(
1206    form: CreateInboxMessageForm, conn: database_dependency, _: AdminAuthDep
1207) -> Inbox:
1208    """Create an inbox message.
1209
1210    Args:
1211        form: form with the message details
1212        conn: database connection
1213
1214    Returns:
1215        created inbox message
1216
1217    Raises:
1218        HTTPException: if failed to create message
1219    """
1220    created_message = await InboxQuerier(conn).create_inbox_message(
1221        CreateInboxMessageParams(
1222            user_id=form.user_id,
1223            sender_id=form.sender_id,
1224            message_subject=form.message_subject,
1225            message_text=form.message_text,
1226        )
1227    )
1228    if not created_message:
1229        raise HTTPException(
1230            status.HTTP_500_INTERNAL_SERVER_ERROR, "Failed to create message"
1231        )
1232    return created_message

Create an inbox message.

Arguments:
  • form: form with the message details
  • conn: database connection
Returns:

created inbox message

Raises:
  • HTTPException: if failed to create message
@router.delete('/database/inbox/{message_id}', status_code=status.HTTP_200_OK, summary='Delete inbox message', description='Deletes a specific inbox message from the system.')
async def delete_inbox_message( message_id: int, 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)], _: Annotated[internal.queries.token.GetSessionByTokenRow, Security(dependency=<function admin_auth>, use_cache=True, scope=None, scopes=None)]) -> internal.queries.models.Inbox: 
1235@router.delete(
1236    "/database/inbox/{message_id}",
1237    status_code=status.HTTP_200_OK,
1238    summary="Delete inbox message",
1239    description="Deletes a specific inbox message from the system.",
1240)
1241async def delete_inbox_message(
1242    message_id: int, conn: database_dependency, _: AdminAuthDep
1243) -> Inbox:
1244    """Delete an inbox message.
1245
1246    Args:
1247        message_id: message id
1248        conn: database connection
1249
1250    Returns:
1251        deleted message
1252
1253    Raises:
1254        HTTPException: if message not found
1255    """
1256    deleted_message = await InboxQuerier(conn).delete_inbox_message(
1257        message_id=message_id
1258    )
1259    if not deleted_message:
1260        raise HTTPException(status.HTTP_404_NOT_FOUND, "Message not found")
1261    return deleted_message

Delete an inbox message.

Arguments:
  • message_id: message id
  • conn: database connection
Returns:

deleted message

Raises:
  • HTTPException: if message not found