Skip to content

Commit

Permalink
chore: Run ruff check for FAST002
Browse files Browse the repository at this point in the history
  • Loading branch information
jpmckinney committed Nov 25, 2024
1 parent 59cfbda commit 4b9ce9f
Show file tree
Hide file tree
Showing 8 changed files with 333 additions and 247 deletions.
27 changes: 17 additions & 10 deletions app/dependencies.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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.
Expand All @@ -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.
Expand All @@ -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,
Expand Down Expand Up @@ -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))
Expand All @@ -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
Expand Down Expand Up @@ -193,24 +198,26 @@ 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

return inner


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)


Expand Down
203 changes: 113 additions & 90 deletions app/routers/applications.py
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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.
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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(
Expand All @@ -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(
Expand All @@ -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.
Expand All @@ -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.
Expand All @@ -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.
Expand Down Expand Up @@ -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.
Expand Down
Loading

0 comments on commit 4b9ce9f

Please sign in to comment.