prepping to rebuild

Author: joeygrable94

app/api/deps/get_client_ip.py

@@ -1,15 +1,17 @@
+from ipaddress import IPv4Address
 from typing import Annotated
 
 from fastapi import Depends, Request
+from pydantic import IPvAnyAddress
 
 
-def get_request_client_ip(request: Request) -> str:
+def get_request_client_ip(request: Request) -> IPv4Address:
     forwarded_ip = request.headers.get("X-Forwarded-For")
     if forwarded_ip:
-        return forwarded_ip.split(",")[0]
+        return IPv4Address(address=forwarded_ip.split(",")[0])
     elif request.client is not None:
-        return request.client.host
-    return "127.0.0.1"  # pragma: no cover
+        return IPv4Address(request.client.host)
+    return IPv4Address("127.0.0.1")  # pragma: no cover
 
 
-RequestClientIp = Annotated[str, Depends(get_request_client_ip)]
+RequestClientIp = Annotated[IPvAnyAddress, Depends(get_request_client_ip)]

app/core/security/auth/exceptions.py

@@ -22,12 +22,11 @@ class HTTPAuth0Error(BaseModel):
 
 
 def configure_authorization_exceptions(app: FastAPI) -> None:
-
     @app.exception_handler(Auth0UnauthenticatedException)
     async def auth0_unauthenticated_exception_handler(
         request: Request, exc: Auth0UnauthenticatedException
     ) -> Response:  # noqa: E501
-        request_headers = {}
+        request_headers: dict = {}
         if exc.headers is not None:
             request_headers.update(exc.headers)  # pragma: no cover
         return await http_exception_handler(
@@ -43,7 +42,7 @@ async def auth0_unauthenticated_exception_handler(
     async def auth0_unauthorized_exception_handler(
         request: Request, exc: Auth0UnauthorizedException
     ) -> Response:  # noqa: E501
-        request_headers = {}
+        request_headers: dict = {}
         if exc.headers is not None:
             request_headers.update(exc.headers)
         return await http_exception_handler(

app/tasks/background.py

@@ -69,7 +69,7 @@ async def bg_task_create_client_data_bucket(
         logger.warning(e)
 
 
-async def bg_task_track_user_ipinfo(ip_address: str, user_id: str) -> None:
+async def bg_task_track_user_ipinfo(ip_address: IPvAnyAddress, user_id: str) -> None:
     """A background task to track the IP Address of a user.
 
     This function will:
@@ -80,7 +80,7 @@ async def bg_task_track_user_ipinfo(ip_address: str, user_id: str) -> None:
     """
     try:
         user_uuid = parse_id(user_id)
-        ip = IPvAnyAddress(ip_address)
+        ip: IPvAnyAddress = ip_address
         ip_in_db: Ipaddress | None = await get_ipaddress_from_db(ip)  # type: ignore
         if ip_in_db is None:
             ip_details: IpinfoResponse = get_ipinfo_details(ip)  # type: ignore

tests/api/v1/clients/test_clients_delete.py

@@ -8,7 +8,7 @@
 
 from app.api.exceptions import ErrorCode
 from app.core.config import settings
-from app.models import User, UserClient
+from app.models import User
 from app.schemas import ClientRead
 
 pytestmark = pytest.mark.asyncio
@@ -43,9 +43,7 @@ async def test_delete_client_by_id_as_client_a(
         db_session=db_session, email=settings.auth.first_client_a
     )
     a_client: ClientRead = await create_random_client(db_session)
-    user_client: UserClient = await assign_user_to_client(  # noqa: F841
-        db_session, client_a, a_client
-    )
+    await assign_user_to_client(db_session, client_a, a_client)
     response: Response = await client.delete(
         f"clients/{a_client.id}",
         headers=client_a_token_headers,