Added annotated fastapi dependencies

.github/workflows/tests.yml

@@ -38,7 +38,6 @@ jobs:
           pip install -r requirements_dev.txt
           pip install -e .
           gunicorn tests.service:__hug_wsgi__ -D
-          uvicorn fastapi_tests.service:app --port 5000 &
           # pre-commit run --all-files  TODO: check what's wrong in CI
           pytest tests
           pytest fastapi_tests

CHANGELOG.rst

@@ -2,6 +2,10 @@
 History
 =======
 
+0.96.1 (2024-09-26)
+-------------------
+* Added annotated fastapi dependencies.
+
 0.96.0 (2024-07-25)
 -------------------
 * Possible fix for Honeybadger exception masking the actual exceptions.

apphelpers/__init__.py

@@ -4,4 +4,4 @@
 
 __author__ = """Scroll Tech"""
 __email__ = "[email protected]"
-__version__ = "0.96.0"
+__version__ = "0.96.1"

apphelpers/rest/fastapi.py

@@ -1,8 +1,9 @@
 import inspect
 from functools import wraps
+from typing import Annotated
 
 from converge import settings
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, Header
 from fastapi.routing import APIRoute
 from starlette.requests import Request
 
@@ -111,7 +112,7 @@ def f_wrapped(*args, **kw):
 
 
 async def get_current_user(request: Request):
-    return request.state.user
+    return request.state.user if request.state.user.id else None
 
 
 async def get_current_user_id(request: Request):
@@ -150,15 +151,21 @@ async def get_user_agent(request: Request):
     return request.headers.get("USER-AGENT", "")
 
 
-user = Depends(get_current_user)
-user_id = Depends(get_current_user_id)
-user_name = Depends(get_current_user_name)
-user_email = Depends(get_current_user_email)
-user_mobile = Depends(get_current_user_mobile)
-domain = Depends(get_current_domain)
-raw_body = Depends(get_raw_body)
-json_body = Depends(get_json_body)
-user_agent = Depends(get_user_agent)
+async def get_user_ip(request: Request):
+    return request.headers["X-FORWARDED-FOR"]
+
+
+user = Annotated[User, Depends(get_current_user)]
+user_id = Annotated[int, Depends(get_current_user_id)]
+user_name = Annotated[str, Depends(get_current_user_name)]
+user_email = Annotated[str, Depends(get_current_user_email)]
+user_mobile = Annotated[str, Depends(get_current_user_mobile)]
+domain = Annotated[str, Depends(get_current_domain)]
+raw_body = Annotated[bytes, Depends(get_raw_body)]
+json_body = Annotated[dict, Depends(get_json_body)]
+user_agent = Annotated[str, Depends(get_user_agent)]
+user_ip = Annotated[str, Depends(get_user_ip)]
+header = Annotated[str, Header()]
 
 
 def dbtransaction(engine, allow_nested=True):

fastapi_tests/app/endpoints.py

@@ -4,36 +4,40 @@
 from pydantic import BaseModel
 
 from apphelpers.rest import endpoint as ep
-from apphelpers.rest.fastapi import json_body, user, user_agent, user_id
+from apphelpers.rest.fastapi import header, json_body, user, user_agent, user_id
 from fastapi_tests.app.models import Book
 
 
-def echo(word, user=user):
+def echo(word, user: user):
     return "%s:%s" % (user.id, word) if user else word
 
 
-async def echo_async(word, user=user):
+async def echo_async(word, user: user):
     return "%s:%s" % (user.id, word) if user else word
 
 
-def echo_post(data=json_body, user=user):
+def echo_post(data: json_body, user: user):
     return "%s:%s" % (user.id, data) if user else data
 
 
+def echo_header(x_key: header):
+    return x_key
+
+
 @ep.login_required
-def secure_echo(word, user=user_id):
-    return "%s:%s" % (user.id, word) if user else word
+def secure_echo(word, user_id: user_id):
+    return "%s:%s" % (user_id, word) if user_id else word
 
 
 @ep.groups_forbidden("noaccess-group")
 @ep.all_groups_required("access-group")
-def echo_groups(user=user):
+def echo_groups(user: user):
     return user.groups
 
 
 @ep.groups_forbidden("noaccess-group")
 @ep.all_groups_required("access-group")
-async def echo_groups_async(user=user):
+async def echo_groups_async(user: user):
     return user.groups
 
 
@@ -42,7 +46,7 @@ def add(nums):
 
 
 @ep.login_required
-def get_my_uid(body=json_body):
+def get_my_uid(body: json_body):
     return body["uid"]
 
 
@@ -62,24 +66,24 @@ async def get_snake_async(name=None):
 
 @ep.all_groups_required("access-group")
 @ep.groups_forbidden("noaccess-group")
-def echo_site_groups(site_id: int, user=user):
+def echo_site_groups(site_id: int, user: user):
     return user.site_groups[site_id]
 
 
 @ep.all_groups_required("access-group")
 @ep.groups_forbidden("noaccess-group")
-async def echo_site_groups_async(site_id: int, user=user):
+async def echo_site_groups_async(site_id: int, user: user):
     return user.site_groups[site_id]
 
 
 @ep.login_required
-async def echo_user_agent_async(user_agent=user_agent):
+async def echo_user_agent_async(user_agent: user_agent):
     return user_agent
 
 
 @ep.login_required
 @ep.ignore_site_ctx
-async def echo_user_agent_without_site_ctx_async(user_agent=user_agent):
+async def echo_user_agent_without_site_ctx_async(user_agent: user_agent):
     return user_agent
 
 
@@ -110,6 +114,7 @@ def setup_routes(factory):
     factory.get("/echo/{word}")(echo)
     factory.get("/echo-async/{word}")(echo_async)
     factory.post("/echo")(echo_post)
+    factory.get("/echo-header")(echo_header)
 
     factory.get("/add")(add)