From 4b9ce9f5a08352da1e0309f6ee68c12eb43a04f6 Mon Sep 17 00:00:00 2001 From: James McKinney <26463+jpmckinney@users.noreply.github.com> Date: Mon, 25 Nov 2024 12:42:48 -0500 Subject: [PATCH] chore: Run ruff check for FAST002 --- app/dependencies.py | 27 ++-- app/routers/applications.py | 203 +++++++++++++++------------ app/routers/downloads.py | 27 ++-- app/routers/guest/applications.py | 220 ++++++++++++++++++------------ app/routers/guest/emails.py | 12 +- app/routers/lenders.py | 24 ++-- app/routers/statistics.py | 23 ++-- app/routers/users.py | 44 +++--- 8 files changed, 333 insertions(+), 247 deletions(-) diff --git a/app/dependencies.py b/app/dependencies.py index 6a678cad..f8f581b2 100644 --- a/app/dependencies.py +++ b/app/dependencies.py @@ -1,7 +1,7 @@ from collections.abc import Callable, Generator from datetime import datetime from enum import Enum -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Annotated, Any from fastapi import Depends, Form, HTTPException, Request, status from sqlalchemy.orm import Session, defaultload, joinedload @@ -30,7 +30,9 @@ async def get_auth_credentials(request: Request) -> auth.JWTAuthorizationCredent return await auth.JWTAuthorization()(request) -async def get_current_user(credentials: auth.JWTAuthorizationCredentials = Depends(get_auth_credentials)) -> Any: +async def get_current_user( + credentials: Annotated[auth.JWTAuthorizationCredentials, Depends(get_auth_credentials)], +) -> Any: """ Extract the username of the current user from the provided JWT credentials. @@ -47,7 +49,9 @@ async def get_current_user(credentials: auth.JWTAuthorizationCredentials = Depen ) from None -async def get_user(username: str = Depends(get_current_user), session: Session = Depends(get_db)) -> models.User: +async def get_user( + username: Annotated[str, Depends(get_current_user)], session: Annotated[Session, Depends(get_db)] +) -> models.User: """ Retrieve the user from the database using the username extracted from the provided JWT credentials. @@ -65,7 +69,7 @@ async def get_user(username: str = Depends(get_current_user), session: Session = return user -async def get_admin_user(user: models.User = Depends(get_user)) -> models.User: +async def get_admin_user(user: Annotated[models.User, Depends(get_user)]) -> models.User: if not user.is_admin(): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, @@ -127,7 +131,7 @@ def raise_if_unauthorized( ) -def get_application_as_user(id: int, session: Session = Depends(get_db)) -> models.Application: +def get_application_as_user(id: int, session: Annotated[Session, Depends(get_db)]) -> models.Application: application = ( models.Application.filter_by(session, "id", id) .options(joinedload(models.Application.borrower), joinedload(models.Application.award)) @@ -149,7 +153,8 @@ def get_scoped_application_as_user( statuses: tuple[models.ApplicationStatus, ...] = (), ) -> Callable[[models.Application, models.User], models.Application]: def inner( - application: models.Application = Depends(get_application_as_user), user: models.User = Depends(get_user) + application: Annotated[models.Application, Depends(get_application_as_user)], + user: Annotated[models.User, Depends(get_user)], ) -> models.Application: raise_if_unauthorized(application, user, roles=roles, scopes=scopes, statuses=statuses) return application @@ -193,7 +198,7 @@ def _get_scoped_application_as_guest_inner( scopes: tuple[ApplicationScope, ...] = (), statuses: tuple[models.ApplicationStatus, ...] = (), ) -> Callable[[models.Application], models.Application]: - def inner(application: models.Application = Depends(depends)) -> models.Application: + def inner(application: Annotated[models.Application, Depends(depends)]) -> models.Application: raise_if_unauthorized(application, scopes=scopes, statuses=statuses) return application @@ -201,16 +206,18 @@ def inner(application: models.Application = Depends(depends)) -> models.Applicat def get_application_as_guest_via_payload( - payload: parsers.ApplicationBase, session: Session = Depends(get_db) + payload: parsers.ApplicationBase, session: Annotated[Session, Depends(get_db)] ) -> models.Application: return _get_application_as_guest_via_uuid(session, payload.uuid) -def get_application_as_guest_via_uuid(uuid: str, session: Session = Depends(get_db)) -> models.Application: +def get_application_as_guest_via_uuid(uuid: str, session: Annotated[Session, Depends(get_db)]) -> models.Application: return _get_application_as_guest_via_uuid(session, uuid) -def get_application_as_guest_via_form(uuid: str = Form(...), session: Session = Depends(get_db)) -> models.Application: +def get_application_as_guest_via_form( + uuid: Annotated[str, Form(...)], session: Annotated[Session, Depends(get_db)] +) -> models.Application: return _get_application_as_guest_via_uuid(session, uuid) diff --git a/app/routers/applications.py b/app/routers/applications.py index 3242fae0..2beb3468 100644 --- a/app/routers/applications.py +++ b/app/routers/applications.py @@ -1,5 +1,5 @@ from datetime import datetime -from typing import Any, cast +from typing import Annotated, Any, cast from fastapi import APIRouter, Depends, HTTPException, Query, status from fastapi.encoders import jsonable_encoder @@ -21,15 +21,18 @@ ) async def reject_application( payload: parsers.LenderRejectedApplication, - session: Session = Depends(get_db), - client: aws.Client = Depends(dependencies.get_aws_client), - user: models.User = Depends(dependencies.get_user), - application: models.Application = Depends( - dependencies.get_scoped_application_as_user( - roles=(models.UserType.FI,), - statuses=(models.ApplicationStatus.STARTED,), - ) - ), + session: Annotated[Session, Depends(get_db)], + client: Annotated[aws.Client, Depends(dependencies.get_aws_client)], + user: Annotated[models.User, Depends(dependencies.get_user)], + application: Annotated[ + models.Application, + Depends( + dependencies.get_scoped_application_as_user( + roles=(models.UserType.FI,), + statuses=(models.ApplicationStatus.STARTED,), + ) + ), + ], ) -> Any: """ Reject an application. @@ -77,15 +80,18 @@ async def reject_application( ) async def approve_application( payload: parsers.LenderApprovedData, - session: Session = Depends(get_db), - client: aws.Client = Depends(dependencies.get_aws_client), - user: models.User = Depends(dependencies.get_user), - application: models.Application = Depends( - dependencies.get_scoped_application_as_user( - roles=(models.UserType.FI,), - statuses=(models.ApplicationStatus.STARTED,), - ) - ), + session: Annotated[Session, Depends(get_db)], + client: Annotated[aws.Client, Depends(dependencies.get_aws_client)], + user: Annotated[models.User, Depends(dependencies.get_user)], + application: Annotated[ + models.Application, + Depends( + dependencies.get_scoped_application_as_user( + roles=(models.UserType.FI,), + statuses=(models.ApplicationStatus.STARTED,), + ) + ), + ], ) -> Any: """ Approve an application. @@ -140,17 +146,20 @@ async def approve_application( ) async def verify_data_field( payload: parsers.UpdateDataField, - session: Session = Depends(get_db), - user: models.User = Depends(dependencies.get_user), - application: models.Application = Depends( - dependencies.get_scoped_application_as_user( - roles=(models.UserType.FI,), - statuses=( - models.ApplicationStatus.STARTED, - models.ApplicationStatus.INFORMATION_REQUESTED, - ), - ) - ), + session: Annotated[Session, Depends(get_db)], + user: Annotated[models.User, Depends(dependencies.get_user)], + application: Annotated[ + models.Application, + Depends( + dependencies.get_scoped_application_as_user( + roles=(models.UserType.FI,), + statuses=( + models.ApplicationStatus.STARTED, + models.ApplicationStatus.INFORMATION_REQUESTED, + ), + ) + ), + ], ) -> Any: """ Verify and update a data field in an application. @@ -190,8 +199,8 @@ async def verify_data_field( async def verify_document( document_id: int, payload: parsers.VerifyBorrowerDocument, - session: Session = Depends(get_db), - user: models.User = Depends(dependencies.get_user), + session: Annotated[Session, Depends(get_db)], + user: Annotated[models.User, Depends(dependencies.get_user)], ) -> Any: """ Verify a borrower document in an application. @@ -231,14 +240,17 @@ async def verify_document( async def update_application_award( id: int, payload: parsers.AwardUpdate, - user: models.User = Depends(dependencies.get_user), - session: Session = Depends(get_db), - application: models.Application = Depends( - dependencies.get_scoped_application_as_user( - roles=(models.UserType.OCP, models.UserType.FI), - statuses=dependencies.USER_CAN_EDIT_AWARD_BORROWER_DATA, - ) - ), + user: Annotated[models.User, Depends(dependencies.get_user)], + session: Annotated[Session, Depends(get_db)], + application: Annotated[ + models.Application, + Depends( + dependencies.get_scoped_application_as_user( + roles=(models.UserType.OCP, models.UserType.FI), + statuses=dependencies.USER_CAN_EDIT_AWARD_BORROWER_DATA, + ) + ), + ], ) -> Any: """ Update the award details of an application. @@ -276,14 +288,17 @@ async def update_application_award( async def update_application_borrower( id: int, payload: parsers.BorrowerUpdate, - user: models.User = Depends(dependencies.get_user), - session: Session = Depends(get_db), - application: models.Application = Depends( - dependencies.get_scoped_application_as_user( - roles=(models.UserType.OCP, models.UserType.FI), - statuses=dependencies.USER_CAN_EDIT_AWARD_BORROWER_DATA, - ) - ), + user: Annotated[models.User, Depends(dependencies.get_user)], + session: Annotated[Session, Depends(get_db)], + application: Annotated[ + models.Application, + Depends( + dependencies.get_scoped_application_as_user( + roles=(models.UserType.OCP, models.UserType.FI), + statuses=dependencies.USER_CAN_EDIT_AWARD_BORROWER_DATA, + ) + ), + ], ) -> Any: """ Update the borrower details of an application. @@ -325,13 +340,13 @@ async def update_application_borrower( tags=[util.Tags.applications], ) async def get_applications_list( - page: int = Query(0, ge=0), - page_size: int = Query(10, gt=0), - sort_field: str = Query("application.borrower_submitted_at"), - sort_order: SortOrder = Query("asc"), - search_value: str = "", - admin: models.User = Depends(dependencies.get_admin_user), - session: Session = Depends(get_db), + page: Annotated[int, Query(default=0, ge=0)], + page_size: Annotated[int, Query(default=10, gt=0)], + sort_field: Annotated[str, Query(default="application.borrower_submitted_at")], + sort_order: Annotated[SortOrder, Query(default="asc")], + search_value: Annotated[str, Query(default="")], + admin: Annotated[models.User, Depends(dependencies.get_admin_user)], + session: Annotated[Session, Depends(get_db)], ) -> serializers.ApplicationListResponse: """Get a paginated list of submitted applications for administrative purposes.""" applications_query = models.Application.submitted_search( @@ -355,13 +370,13 @@ async def get_applications_list( tags=[util.Tags.applications], ) async def get_applications( - page: int = Query(0, ge=0), - page_size: int = Query(10, gt=0), - sort_field: str = Query("application.borrower_submitted_at"), - sort_order: SortOrder = Query("asc"), - search_value: str = "", - user: models.User = Depends(dependencies.get_user), - session: Session = Depends(get_db), + page: Annotated[int, Query(default=0, ge=0)], + page_size: Annotated[int, Query(default=10, gt=0)], + sort_field: Annotated[str, Query(default="application.borrower_submitted_at")], + sort_order: Annotated[SortOrder, Query(default="asc")], + search_value: Annotated[str, Query(default="")], + user: Annotated[models.User, Depends(dependencies.get_user)], + session: Annotated[Session, Depends(get_db)], ) -> serializers.ApplicationListResponse: """Get a paginated list of submitted applications for a specific lender user.""" applications_query = models.Application.submitted_search( @@ -386,11 +401,12 @@ async def get_applications( response_model=models.ApplicationWithRelations, ) async def get_application( - user: models.User = Depends(dependencies.get_user), - session: Session = Depends(get_db), - application: models.Application = Depends( - dependencies.get_scoped_application_as_user(roles=(models.UserType.OCP, models.UserType.FI)) - ), + user: Annotated[models.User, Depends(dependencies.get_user)], + session: Annotated[Session, Depends(get_db)], + application: Annotated[ + models.Application, + Depends(dependencies.get_scoped_application_as_user(roles=(models.UserType.OCP, models.UserType.FI))), + ], ) -> Any: """ Retrieve an application by its ID. @@ -408,14 +424,17 @@ async def get_application( ) async def start_application( id: int, - user: models.User = Depends(dependencies.get_user), - session: Session = Depends(get_db), - application: models.Application = Depends( - dependencies.get_scoped_application_as_user( - roles=(models.UserType.FI,), - statuses=(models.ApplicationStatus.SUBMITTED,), - ) - ), + user: Annotated[models.User, Depends(dependencies.get_user)], + session: Annotated[Session, Depends(get_db)], + application: Annotated[ + models.Application, + Depends( + dependencies.get_scoped_application_as_user( + roles=(models.UserType.FI,), + statuses=(models.ApplicationStatus.SUBMITTED,), + ) + ), + ], ) -> Any: """ Start an application. @@ -441,16 +460,19 @@ async def start_application( ) async def email_borrower( payload: parsers.ApplicationEmailBorrower, - session: Session = Depends(get_db), - client: aws.Client = Depends(dependencies.get_aws_client), - user: models.User = Depends(dependencies.get_user), - application: models.Application = Depends( - dependencies.get_scoped_application_as_user( - roles=(models.UserType.FI,), - scopes=(dependencies.ApplicationScope.NATIVE,), - statuses=(models.ApplicationStatus.STARTED,), - ) - ), + session: Annotated[Session, Depends(get_db)], + client: Annotated[aws.Client, Depends(dependencies.get_aws_client)], + user: Annotated[models.User, Depends(dependencies.get_user)], + application: Annotated[ + models.Application, + Depends( + dependencies.get_scoped_application_as_user( + roles=(models.UserType.FI,), + scopes=(dependencies.ApplicationScope.NATIVE,), + statuses=(models.ApplicationStatus.STARTED,), + ) + ), + ], ) -> Any: """ Send an email to the borrower and update the application status. @@ -492,11 +514,12 @@ async def email_borrower( tags=[util.Tags.applications], ) async def previous_contracts( - user: models.User = Depends(dependencies.get_user), - session: Session = Depends(get_db), - application: models.Application = Depends( - dependencies.get_scoped_application_as_user(roles=(models.UserType.OCP, models.UserType.FI)) - ), + user: Annotated[models.User, Depends(dependencies.get_user)], + session: Annotated[Session, Depends(get_db)], + application: Annotated[ + models.Application, + Depends(dependencies.get_scoped_application_as_user(roles=(models.UserType.OCP, models.UserType.FI))), + ], ) -> list[models.Award]: """ Get the previous awards associated with an application. diff --git a/app/routers/downloads.py b/app/routers/downloads.py index 51da56c6..0fa3f560 100644 --- a/app/routers/downloads.py +++ b/app/routers/downloads.py @@ -1,7 +1,7 @@ import csv import io import zipfile -from typing import Any +from typing import Annotated, Any from fastapi import APIRouter, Depends, Response from reportlab.lib.pagesizes import letter @@ -24,8 +24,8 @@ ) async def get_borrower_document( document_id: int, - session: Session = Depends(get_db), - user: models.User = Depends(dependencies.get_user), + session: Annotated[Session, Depends(get_db)], + user: Annotated[models.User, Depends(dependencies.get_user)], ) -> Response: """ Retrieve a borrower document by its ID and stream the file content as a response. @@ -62,13 +62,16 @@ async def get_borrower_document( ) async def download_application( lang: str, - session: Session = Depends(get_db), - user: models.User = Depends(dependencies.get_user), - application: models.Application = Depends( - dependencies.get_scoped_application_as_user( - roles=(models.UserType.OCP, models.UserType.FI), scopes=(ApplicationScope.UNEXPIRED,) - ) - ), + session: Annotated[Session, Depends(get_db)], + user: Annotated[models.User, Depends(dependencies.get_user)], + application: Annotated[ + models.Application, + Depends( + dependencies.get_scoped_application_as_user( + roles=(models.UserType.OCP, models.UserType.FI), scopes=(ApplicationScope.UNEXPIRED,) + ) + ), + ], ) -> Response: """ Retrieve all documents related to an application and stream them as a zip file. @@ -137,8 +140,8 @@ async def download_application( ) async def export_applications( lang: str, - user: models.User = Depends(dependencies.get_user), - session: Session = Depends(get_db), + user: Annotated[models.User, Depends(dependencies.get_user)], + session: Annotated[Session, Depends(get_db)], ) -> Response: stream = io.StringIO() diff --git a/app/routers/guest/applications.py b/app/routers/guest/applications.py index 6744adf6..e6743c43 100644 --- a/app/routers/guest/applications.py +++ b/app/routers/guest/applications.py @@ -1,5 +1,5 @@ from datetime import datetime -from typing import Any, cast +from typing import Annotated, Any, cast from fastapi import APIRouter, BackgroundTasks, Depends, Form, HTTPException, UploadFile, status from fastapi.encoders import jsonable_encoder @@ -19,10 +19,13 @@ tags=[util.Tags.applications], ) async def application_by_uuid( - session: Session = Depends(get_db), - application: models.Application = Depends( - dependencies.get_scoped_application_as_guest_via_uuid(scopes=(dependencies.ApplicationScope.UNEXPIRED,)) - ), + session: Annotated[Session, Depends(get_db)], + application: Annotated[ + models.Application, + Depends( + dependencies.get_scoped_application_as_guest_via_uuid(scopes=(dependencies.ApplicationScope.UNEXPIRED,)) + ), + ], ) -> serializers.ApplicationResponse: """ Retrieve an application by its UUID. @@ -46,12 +49,15 @@ async def application_by_uuid( ) async def decline( payload: parsers.ApplicationDeclinePayload, - session: Session = Depends(get_db), - application: models.Application = Depends( - dependencies.get_scoped_application_as_guest_via_payload( - scopes=(dependencies.ApplicationScope.UNEXPIRED,), statuses=(models.ApplicationStatus.PENDING,) - ) - ), + session: Annotated[Session, Depends(get_db)], + application: Annotated[ + models.Application, + Depends( + dependencies.get_scoped_application_as_guest_via_payload( + scopes=(dependencies.ApplicationScope.UNEXPIRED,), statuses=(models.ApplicationStatus.PENDING,) + ) + ), + ], ) -> serializers.ApplicationResponse: """ Decline an application. @@ -91,12 +97,15 @@ async def decline( ) async def rollback_decline( payload: parsers.ApplicationBase, - session: Session = Depends(get_db), - application: models.Application = Depends( - dependencies.get_scoped_application_as_guest_via_payload( - scopes=(dependencies.ApplicationScope.UNEXPIRED,), statuses=(models.ApplicationStatus.DECLINED,) - ) - ), + session: Annotated[Session, Depends(get_db)], + application: Annotated[ + models.Application, + Depends( + dependencies.get_scoped_application_as_guest_via_payload( + scopes=(dependencies.ApplicationScope.UNEXPIRED,), statuses=(models.ApplicationStatus.DECLINED,) + ) + ), + ], ) -> serializers.ApplicationResponse: """ Rollback the decline of an application. @@ -130,12 +139,15 @@ async def rollback_decline( ) async def decline_feedback( payload: parsers.ApplicationDeclineFeedbackPayload, - session: Session = Depends(get_db), - application: models.Application = Depends( - dependencies.get_scoped_application_as_guest_via_payload( - scopes=(dependencies.ApplicationScope.UNEXPIRED,), statuses=(models.ApplicationStatus.DECLINED,) - ) - ), + session: Annotated[Session, Depends(get_db)], + application: Annotated[ + models.Application, + Depends( + dependencies.get_scoped_application_as_guest_via_payload( + scopes=(dependencies.ApplicationScope.UNEXPIRED,), statuses=(models.ApplicationStatus.DECLINED,) + ) + ), + ], ) -> serializers.ApplicationResponse: """ Provide feedback for a declined application. @@ -165,12 +177,15 @@ async def decline_feedback( async def access_scheme( payload: parsers.ApplicationBase, background_tasks: BackgroundTasks, - session: Session = Depends(get_db), - application: models.Application = Depends( - dependencies.get_scoped_application_as_guest_via_payload( - scopes=(dependencies.ApplicationScope.UNEXPIRED,), statuses=(models.ApplicationStatus.PENDING,) - ) - ), + session: Annotated[Session, Depends(get_db)], + application: Annotated[ + models.Application, + Depends( + dependencies.get_scoped_application_as_guest_via_payload( + scopes=(dependencies.ApplicationScope.UNEXPIRED,), statuses=(models.ApplicationStatus.PENDING,) + ) + ), + ], ) -> serializers.ApplicationResponse: """ Access the scheme for an application. @@ -205,10 +220,13 @@ async def access_scheme( ) async def credit_product_options( payload: parsers.ApplicationCreditOptions, - session: Session = Depends(get_db), - application: models.Application = Depends( - dependencies.get_scoped_application_as_guest_via_payload(statuses=(models.ApplicationStatus.ACCEPTED,)) - ), + session: Annotated[Session, Depends(get_db)], + application: Annotated[ + models.Application, + Depends( + dependencies.get_scoped_application_as_guest_via_payload(statuses=(models.ApplicationStatus.ACCEPTED,)) + ), + ], ) -> serializers.CreditProductListResponse: """ Get the available credit product options for an application. @@ -243,10 +261,13 @@ async def credit_product_options( ) async def select_credit_product( payload: parsers.ApplicationSelectCreditProduct, - session: Session = Depends(get_db), - application: models.Application = Depends( - dependencies.get_scoped_application_as_guest_via_payload(statuses=(models.ApplicationStatus.ACCEPTED,)) - ), + session: Annotated[Session, Depends(get_db)], + application: Annotated[ + models.Application, + Depends( + dependencies.get_scoped_application_as_guest_via_payload(statuses=(models.ApplicationStatus.ACCEPTED,)) + ), + ], ) -> serializers.ApplicationResponse: """ Select a credit product for an application. @@ -299,10 +320,13 @@ async def select_credit_product( ) async def rollback_select_credit_product( payload: parsers.ApplicationBase, - session: Session = Depends(get_db), - application: models.Application = Depends( - dependencies.get_scoped_application_as_guest_via_payload(statuses=(models.ApplicationStatus.ACCEPTED,)) - ), + session: Annotated[Session, Depends(get_db)], + application: Annotated[ + models.Application, + Depends( + dependencies.get_scoped_application_as_guest_via_payload(statuses=(models.ApplicationStatus.ACCEPTED,)) + ), + ], ) -> serializers.ApplicationResponse: """ Rollback the selection of a credit product for an application. @@ -348,10 +372,13 @@ async def rollback_select_credit_product( ) async def confirm_credit_product( payload: parsers.ApplicationBase, - session: Session = Depends(get_db), - application: models.Application = Depends( - dependencies.get_scoped_application_as_guest_via_payload(statuses=(models.ApplicationStatus.ACCEPTED,)) - ), + session: Annotated[Session, Depends(get_db)], + application: Annotated[ + models.Application, + Depends( + dependencies.get_scoped_application_as_guest_via_payload(statuses=(models.ApplicationStatus.ACCEPTED,)) + ), + ], ) -> serializers.ApplicationResponse: """ Confirm the selected credit product for an application. @@ -437,10 +464,13 @@ async def confirm_credit_product( ) async def rollback_confirm_credit_product( payload: parsers.ApplicationBase, - session: Session = Depends(get_db), - application: models.Application = Depends( - dependencies.get_scoped_application_as_guest_via_payload(statuses=(models.ApplicationStatus.ACCEPTED,)) - ), + session: Annotated[Session, Depends(get_db)], + application: Annotated[ + models.Application, + Depends( + dependencies.get_scoped_application_as_guest_via_payload(statuses=(models.ApplicationStatus.ACCEPTED,)) + ), + ], ) -> serializers.ApplicationResponse: """ Rollback the confirmation of the selected credit product for an application. @@ -498,11 +528,14 @@ async def rollback_confirm_credit_product( ) async def update_apps_send_notifications( payload: parsers.ApplicationBase, - session: Session = Depends(get_db), - client: aws.Client = Depends(dependencies.get_aws_client), - application: models.Application = Depends( - dependencies.get_scoped_application_as_guest_via_payload(statuses=(models.ApplicationStatus.ACCEPTED,)) - ), + session: Annotated[Session, Depends(get_db)], + client: Annotated[aws.Client, Depends(dependencies.get_aws_client)], + application: Annotated[ + models.Application, + Depends( + dependencies.get_scoped_application_as_guest_via_payload(statuses=(models.ApplicationStatus.ACCEPTED,)) + ), + ], ) -> serializers.ApplicationResponse: """ Change application status from "'ACCEPTED" to "SUBMITTED". Send a notification to OCP and lender user. @@ -551,17 +584,20 @@ async def update_apps_send_notifications( ) async def upload_document( file: UploadFile, - type: str = Form(...), - session: Session = Depends(get_db), - application: models.Application = Depends( - dependencies.get_scoped_application_as_guest_via_form( - statuses=( - models.ApplicationStatus.ACCEPTED, - models.ApplicationStatus.INFORMATION_REQUESTED, - models.ApplicationStatus.APPROVED, + type: Annotated[str, Form(...)], + session: Annotated[Session, Depends(get_db)], + application: Annotated[ + models.Application, + Depends( + dependencies.get_scoped_application_as_guest_via_form( + statuses=( + models.ApplicationStatus.ACCEPTED, + models.ApplicationStatus.INFORMATION_REQUESTED, + models.ApplicationStatus.APPROVED, + ) ) - ) - ), + ), + ], ) -> Any: """ Upload a document for an application. @@ -599,13 +635,16 @@ async def upload_document( ) async def complete_information_request( payload: parsers.ApplicationBase, - client: aws.Client = Depends(dependencies.get_aws_client), - session: Session = Depends(get_db), - application: models.Application = Depends( - dependencies.get_scoped_application_as_guest_via_payload( - statuses=(models.ApplicationStatus.INFORMATION_REQUESTED,) - ) - ), + client: Annotated[aws.Client, Depends(dependencies.get_aws_client)], + session: Annotated[Session, Depends(get_db)], + application: Annotated[ + models.Application, + Depends( + dependencies.get_scoped_application_as_guest_via_payload( + statuses=(models.ApplicationStatus.INFORMATION_REQUESTED,) + ) + ), + ], ) -> serializers.ApplicationResponse: """ Complete the information request for an application. @@ -646,11 +685,14 @@ async def complete_information_request( ) async def find_alternative_credit_option( payload: parsers.ApplicationBase, - session: Session = Depends(get_db), - client: aws.Client = Depends(dependencies.get_aws_client), - application: models.Application = Depends( - dependencies.get_scoped_application_as_guest_via_payload(statuses=(models.ApplicationStatus.REJECTED,)) - ), + session: Annotated[Session, Depends(get_db)], + client: Annotated[aws.Client, Depends(dependencies.get_aws_client)], + application: Annotated[ + models.Application, + Depends( + dependencies.get_scoped_application_as_guest_via_payload(statuses=(models.ApplicationStatus.REJECTED,)) + ), + ], ) -> serializers.ApplicationResponse: """ Find an alternative credit option for a rejected application by copying it. @@ -723,12 +765,15 @@ async def find_alternative_credit_option( tags=[util.Tags.applications], ) async def access_external_onboarding( - session: Session = Depends(get_db), - application: models.Application = Depends( - dependencies.get_scoped_application_as_guest_via_uuid( - statuses=(models.ApplicationStatus.SUBMITTED, models.ApplicationStatus.STARTED) + session: Annotated[Session, Depends(get_db)], + application: Annotated[ + models.Application, + Depends( + dependencies.get_scoped_application_as_guest_via_uuid( + statuses=(models.ApplicationStatus.SUBMITTED, models.ApplicationStatus.STARTED) + ), ), - ), + ], ) -> RedirectResponse: """ :return: A redirect to the lender.external_onboarding_url. @@ -742,12 +787,15 @@ async def access_external_onboarding( tags=[util.Tags.applications], ) async def accessed_external_onboarding( - session: Session = Depends(get_db), - application: models.Application = Depends( - dependencies.get_scoped_application_as_guest_via_uuid( - statuses=(models.ApplicationStatus.SUBMITTED, models.ApplicationStatus.STARTED) + session: Annotated[Session, Depends(get_db)], + application: Annotated[ + models.Application, + Depends( + dependencies.get_scoped_application_as_guest_via_uuid( + statuses=(models.ApplicationStatus.SUBMITTED, models.ApplicationStatus.STARTED) + ), ), - ), + ], ) -> RedirectResponse: """ Report having started external onboarding. diff --git a/app/routers/guest/emails.py b/app/routers/guest/emails.py index 65bb5059..3a49fdfa 100644 --- a/app/routers/guest/emails.py +++ b/app/routers/guest/emails.py @@ -1,3 +1,5 @@ +from typing import Annotated + from fastapi import APIRouter, Depends, HTTPException, status from fastapi.encoders import jsonable_encoder from sqlalchemy.orm import Session @@ -15,9 +17,9 @@ ) async def change_email( payload: parsers.ChangeEmail, - session: Session = Depends(get_db), - client: aws.Client = Depends(dependencies.get_aws_client), - application: models.Application = Depends(dependencies.get_application_as_guest_via_payload), + session: Annotated[Session, Depends(get_db)], + client: Annotated[aws.Client, Depends(dependencies.get_aws_client)], + application: Annotated[models.Application, Depends(dependencies.get_application_as_guest_via_payload)], ) -> parsers.ChangeEmail: """ Change the email address for an application. @@ -64,8 +66,8 @@ async def change_email( ) async def confirm_email( payload: parsers.ConfirmNewEmail, - session: Session = Depends(get_db), - application: models.Application = Depends(dependencies.get_application_as_guest_via_payload), + session: Annotated[Session, Depends(get_db)], + application: Annotated[models.Application, Depends(dependencies.get_application_as_guest_via_payload)], ) -> parsers.ChangeEmail: """ Confirm the email address change for an application. diff --git a/app/routers/lenders.py b/app/routers/lenders.py index 0799b319..dc01e00e 100644 --- a/app/routers/lenders.py +++ b/app/routers/lenders.py @@ -1,4 +1,4 @@ -from typing import Any +from typing import Annotated, Any from fastapi import APIRouter, Depends, HTTPException, status from fastapi.encoders import jsonable_encoder @@ -20,8 +20,8 @@ ) async def create_lender( payload: models.LenderCreate, - admin: models.User = Depends(dependencies.get_admin_user), - session: Session = Depends(get_db), + admin: Annotated[models.User, Depends(dependencies.get_admin_user)], + session: Annotated[Session, Depends(get_db)], ) -> models.Lender: """ Create a new lender. @@ -57,8 +57,8 @@ async def create_lender( async def create_credit_products( lender_id: int, payload: models.CreditProduct, - admin: models.User = Depends(dependencies.get_admin_user), - session: Session = Depends(get_db), + admin: Annotated[models.User, Depends(dependencies.get_admin_user)], + session: Annotated[Session, Depends(get_db)], ) -> models.CreditProduct: """ Create a new credit product for a specific lender. @@ -83,7 +83,7 @@ async def create_credit_products( ) async def get_lender( lender_id: int, - session: Session = Depends(get_db), + session: Annotated[Session, Depends(get_db)], ) -> Any: """ Retrieve a lender by its ID. @@ -102,8 +102,8 @@ async def get_lender( async def update_lender( lender_id: int, payload: models.LenderBase, - admin: models.User = Depends(dependencies.get_admin_user), - session: Session = Depends(get_db), + admin: Annotated[models.User, Depends(dependencies.get_admin_user)], + session: Annotated[Session, Depends(get_db)], ) -> models.Lender: """ Update an existing lender. @@ -132,7 +132,7 @@ async def update_lender( tags=[util.Tags.lenders], ) async def get_lenders_list( - session: Session = Depends(get_db), + session: Annotated[Session, Depends(get_db)], ) -> serializers.LenderListResponse: """ Get the list of all lenders. @@ -170,7 +170,7 @@ async def get_procurement_categories_from_source() -> list[str]: ) async def get_credit_product( credit_product_id: int, - session: Session = Depends(get_db), + session: Annotated[Session, Depends(get_db)], ) -> Any: """ Retrieve a credit product by its ID, including its associated lender information. @@ -201,8 +201,8 @@ async def get_credit_product( async def update_credit_products( credit_product_id: int, payload: models.CreditProduct, - admin: models.User = Depends(dependencies.get_admin_user), - session: Session = Depends(get_db), + admin: Annotated[models.User, Depends(dependencies.get_admin_user)], + session: Annotated[Session, Depends(get_db)], ) -> models.CreditProduct: """ Update an existing credit product. diff --git a/app/routers/statistics.py b/app/routers/statistics.py index 379a41e9..a2f25240 100644 --- a/app/routers/statistics.py +++ b/app/routers/statistics.py @@ -1,6 +1,7 @@ from datetime import datetime, timedelta +from typing import Annotated -from fastapi import APIRouter, Depends +from fastapi import APIRouter, Depends, Query from sqlalchemy.orm import Session import app.utils.statistics as statistics_utils @@ -17,12 +18,12 @@ tags=[util.Tags.statistics], ) async def get_admin_statistics_by_lender( - initial_date: str | None = None, - final_date: str | None = None, - lender_id: int | None = None, - custom_range: StatisticRange | None = None, - admin: User = Depends(dependencies.get_admin_user), - session: Session = Depends(get_db), + initial_date: Annotated[str | None, Query(default=None)], + final_date: Annotated[str | None, Query(default=None)], + lender_id: Annotated[int | None, Query(default=None)], + custom_range: Annotated[StatisticRange | None, Query(default=None)], + admin: Annotated[User, Depends(dependencies.get_admin_user)], + session: Annotated[Session, Depends(get_db)], ) -> serializers.StatisticResponse: """ Retrieve OCP statistics by lender. @@ -62,8 +63,8 @@ async def get_admin_statistics_by_lender( tags=[util.Tags.statistics], ) async def get_admin_statistics_opt_in( - admin: User = Depends(dependencies.get_admin_user), - session: Session = Depends(get_db), + admin: Annotated[User, Depends(dependencies.get_admin_user)], + session: Annotated[Session, Depends(get_db)], ) -> serializers.StatisticOptInResponse: """ Retrieve OCP statistics for borrower opt-in. @@ -83,8 +84,8 @@ async def get_admin_statistics_opt_in( tags=[util.Tags.statistics], ) async def get_lender_statistics( - session: Session = Depends(get_db), - user: User = Depends(dependencies.get_user), + session: Annotated[Session, Depends(get_db)], + user: Annotated[User, Depends(dependencies.get_user)], ) -> serializers.StatisticResponse: """ Retrieve statistics for a lender. diff --git a/app/routers/users.py b/app/routers/users.py index bd4ded23..630163bb 100644 --- a/app/routers/users.py +++ b/app/routers/users.py @@ -1,3 +1,5 @@ +from typing import Annotated + from fastapi import APIRouter, Depends, HTTPException, Query, Request, status from fastapi.encoders import jsonable_encoder from sqlalchemy.exc import IntegrityError @@ -18,9 +20,9 @@ ) async def create_user( payload: models.UserBase, - session: Session = Depends(get_db), - client: aws.Client = Depends(dependencies.get_aws_client), - admin: models.User = Depends(dependencies.get_admin_user), + session: Annotated[Session, Depends(get_db)], + client: Annotated[aws.Client, Depends(dependencies.get_aws_client)], + admin: Annotated[models.User, Depends(dependencies.get_admin_user)], ) -> models.User: """ Create a new user in Cognito. @@ -65,7 +67,7 @@ async def create_user( ) def change_password( payload: parsers.BasicUser, - client: aws.Client = Depends(dependencies.get_aws_client), + client: Annotated[aws.Client, Depends(dependencies.get_aws_client)], ) -> serializers.ChangePasswordResponse | serializers.ResponseBase: """ Change user password. @@ -111,7 +113,7 @@ def change_password( ) def setup_mfa( payload: parsers.SetupMFA, - client: aws.Client = Depends(dependencies.get_aws_client), + client: Annotated[aws.Client, Depends(dependencies.get_aws_client)], ) -> serializers.ResponseBase: """Set up multi-factor authentication (MFA) for the user.""" try: @@ -137,8 +139,8 @@ def setup_mfa( ) def login( payload: parsers.BasicUser, - client: aws.Client = Depends(dependencies.get_aws_client), - session: Session = Depends(get_db), + client: Annotated[aws.Client, Depends(dependencies.get_aws_client)], + session: Annotated[Session, Depends(get_db)], ) -> serializers.LoginResponse: """ Authenticate the user with Multi-Factor Authentication (MFA) and generate access and refresh tokens. @@ -198,7 +200,7 @@ def login( ) async def logout( request: Request, - client: aws.Client = Depends(dependencies.get_aws_client), + client: Annotated[aws.Client, Depends(dependencies.get_aws_client)], ) -> serializers.ResponseBase: """Logout the user from all devices in Cognito.""" try: @@ -220,8 +222,8 @@ async def logout( tags=[util.Tags.authentication], ) def me( - username_from_token: str = Depends(dependencies.get_current_user), - session: Session = Depends(get_db), + username_from_token: Annotated[str, Depends(dependencies.get_current_user)], + session: Annotated[Session, Depends(get_db)], ) -> serializers.UserResponse: """ Get the details of the currently authenticated user. @@ -240,7 +242,7 @@ def me( ) def forgot_password( payload: parsers.ResetPassword, - client: aws.Client = Depends(dependencies.get_aws_client), + client: Annotated[aws.Client, Depends(dependencies.get_aws_client)], ) -> serializers.ResponseBase: """ Initiate the process of resetting a user's password. @@ -269,8 +271,8 @@ def forgot_password( ) async def get_user( user_id: int, - admin: models.User = Depends(dependencies.get_admin_user), - session: Session = Depends(get_db), + admin: Annotated[models.User, Depends(dependencies.get_admin_user)], + session: Annotated[Session, Depends(get_db)], ) -> models.User: """ Retrieve information about a user. @@ -285,12 +287,12 @@ async def get_user( tags=[util.Tags.users], ) async def get_all_users( - page: int = Query(0, ge=0), - page_size: int = Query(10, gt=0), - sort_field: str = Query("created_at"), - sort_order: SortOrder = Query("asc"), - admin: models.User = Depends(dependencies.get_admin_user), - session: Session = Depends(get_db), + page: Annotated[int, Query(default=0, ge=0)], + page_size: Annotated[int, Query(default=10, gt=0)], + sort_field: Annotated[str, Query(default="created_at")], + sort_order: Annotated[SortOrder, Query(default="asc")], + admin: Annotated[models.User, Depends(dependencies.get_admin_user)], + session: Annotated[Session, Depends(get_db)], ) -> serializers.UserListResponse: """ Retrieve a list of users. @@ -324,8 +326,8 @@ async def get_all_users( async def update_user( user_id: int, payload: models.User, - admin: models.User = Depends(dependencies.get_admin_user), - session: Session = Depends(get_db), + admin: Annotated[models.User, Depends(dependencies.get_admin_user)], + session: Annotated[Session, Depends(get_db)], ) -> models.User: """ Update a user's information.