Added annotated fastapi dependencies

.github/workflows/tests.yml

@@ -29,16 +29,13 @@ jobs:
       - name: Setup Python
         uses: actions/setup-python@v2
         with:
-          python-version: 3.8
+          python-version: 3.12
           cache: "pip"
           cache-dependency-path: "**/requirements_dev.txt"
       - name: Run tests
         run: |
           export SETTINGS_DIR=.
           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/db/piccolo.py

@@ -1,9 +1,17 @@
 from contextlib import asynccontextmanager
 from functools import wraps
-from typing import List, Set, Type
+from typing import (
+    List,
+    Set,
+    Type,
+)
 
 from piccolo.engine.postgres import PostgresEngine
-from piccolo.table import Table, create_db_tables_sync, drop_db_tables_sync
+from piccolo.table import (
+    Table,
+    create_db_tables_sync,
+    drop_db_tables_sync,
+)
 
 
 @asynccontextmanager

apphelpers/rest/common.py

@@ -1,7 +1,15 @@
 from __future__ import annotations
 
-from dataclasses import asdict, dataclass, field
-from typing import Dict, List, Optional
+from dataclasses import (
+    asdict,
+    dataclass,
+    field,
+)
+from typing import (
+    Dict,
+    List,
+    Optional,
+)
 
 from converge import settings
 from requests.exceptions import HTTPError

apphelpers/rest/fastapi.py

@@ -1,8 +1,13 @@
 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
 
@@ -15,7 +20,11 @@
     InvalidSessionError,
 )
 from apphelpers.rest import endpoint as ep
-from apphelpers.rest.common import User, notify_honeybadger, phony
+from apphelpers.rest.common import (
+    User,
+    notify_honeybadger,
+    phony,
+)
 from apphelpers.sessions import SessionDBHandler
 
 if settings.get("HONEYBADGER_API_KEY"):
@@ -111,7 +120,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 +159,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):

apphelpers/rest/hug.py

@@ -4,7 +4,11 @@
 
 import hug
 from converge import settings
-from falcon import HTTPForbidden, HTTPNotFound, HTTPUnauthorized
+from falcon import (
+    HTTPForbidden,
+    HTTPNotFound,
+    HTTPUnauthorized,
+)
 from hug.decorators import wraps
 
 from apphelpers.db import dbtransaction

fastapi_tests/app/endpoints.py

@@ -4,36 +4,46 @@
 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 +52,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 +72,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 +120,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)
 

fastapi_tests/app/models.py

@@ -1,7 +1,7 @@
-import settings
 from piccolo import columns as col
 from piccolo.engine.postgres import PostgresEngine
 
+import settings
 from apphelpers.db.piccolo import BaseTable
 
 db = PostgresEngine(

fastapi_tests/service.py

@@ -6,7 +6,6 @@
 sys.path.append(os.getcwd() + "/..")
 
 import settings
-
 from apphelpers.rest.fastapi import APIFactory
 from fastapi_tests.app.endpoints import setup_routes
 from fastapi_tests.app.models import db