dmontagu / fastapi-utils

Reusable utilities for FastAPI

Reusable utilities for FastAPI

---

Documentation: https://fastapi-utils.davidmontague.xyz

Source Code: https://github.com/dmontagu/fastapi-utils

---

FastAPI is a modern, fast web framework for building APIs with Python 3.7+.

But if you're here, you probably already knew that!

---

Features

This package includes a number of utilities to help reduce boilerplate and reuse common functionality across projects:

• Class Based Views: Stop repeating the same dependencies over and over in the signature of related endpoints.
• Response-Model Inferring Router: Let FastAPI infer the response_model to use based on your return type annotation.
• Repeated Tasks: Easily trigger periodic tasks on server startup
• Timing Middleware: Log basic timing information for every request
• SQLAlchemy Sessions: The FastAPISessionMaker class provides an easily-customized SQLAlchemy Session dependency
• OpenAPI Spec Simplification: Simplify your OpenAPI Operation IDs for cleaner output from OpenAPI Generator

---

It also adds a variety of more basic utilities that are useful across a wide variety of projects:

• APIModel: A reusable pydantic.BaseModel-derived base class with useful defaults
• APISettings: A subclass of pydantic.BaseSettings that makes it easy to configure FastAPI through environment variables
• String-Valued Enums: The StrEnum and CamelStrEnum classes make string-valued enums easier to maintain
• CamelCase Conversions: Convenience functions for converting strings from snake_case to camelCase or PascalCase and back
• GUID Type: The provided GUID type makes it easy to use UUIDs as the primary keys for your database tables

See the docs for more details and examples.

Requirements

This package is intended for use with any recent version of FastAPI (depending on pydantic>=1.0), and Python 3.7+.

Installation

pip install fastapi-utils

License

This project is licensed under the terms of the MIT license.

fastapi-utils's Issues

CRUD tools

Hi @dmontagu

Nice work here.

I have developed some CRUD tools, using the FastAPI postgres template "CRUD" base that works together with your CBV. I have also added sorting and filtering using sqlalchemy_filters that can be defined for each CBV.

This should make it much easier to build a basic CRUD app for fast-api.

The filter /sort fields can work through joins too.

This allows for any endpoint to take defined parameters like:
/api/v1/modelname/?limit=100&field_1=eq:something&model2___field_1=eq:somethingelse&sort_by=field1:desc,model2__field1:asc

Basically this includes a CRUD base for the database with each model inheriting from this.
A base CBV that will be used for the standard CRUD stuff.

I think this could work well being in this separate project instead of in the cookiecutter template.

Could you see this fit into the scope of this project?

Thanks,
Jonathan

[UPDATE] Is this project still supported?

Description

Is this project still supported? I see the @dmontagu is still active on GitHub but this project hasn't received any commits in over a year. Would like to know if this is something that is still appropriate for new projects or if one of the forks should be considered instead.

[BUG] Inferring route throws exception when return typehint is set to None.

While it is implied that a function without a return typehint will return None some might choose to explicitly write it in their code. As of right now the following code will raise an unwanted (and likely unintentioned) exception:

from fastapi import FastAPI, status
from fastapi_utils.inferring_router import InferringRouter

app = FastAPI()
router = InferringRouter()

@router.get("/inferred", status_code=status.HTTP_204_NO_CONTENT)
def get_resource(resource_id: int) -> None:
    pass

app.include_router(router)

Error thrown:

  File "./test.py", line 9, in <module>
    def get_resource(resource_id: int) -> None:
  File "/Users/felipe/.pyenv/versions/3.8.5/lib/python3.8/site-packages/fastapi/routing.py", line 551, in decorator
    self.add_api_route(
  File "/Users/felipe/.pyenv/versions/3.8.5/lib/python3.8/site-packages/fastapi_utils/inferring_router.py", line 18, in add_api_route
    return super().add_api_route(path, endpoint, **kwargs)
  File "/Users/felipe/.pyenv/versions/3.8.5/lib/python3.8/site-packages/fastapi/routing.py", line 496, in add_api_route
    route = route_class(
  File "/Users/felipe/.pyenv/versions/3.8.5/lib/python3.8/site-packages/fastapi/routing.py", line 320, in __init__
    assert (
AssertionError: Status code 204 must not have a response body

[BUG] @repeat_every() does not run without @app.on_event('startup') decorator

Describe the bug
The @repeat_every() decorator does not trigger the function it decorates unless the @app.on_event('startup') decorator is also present.

To Reproduce
I created this sample main.py to show the issue I've been seeing.

1. Copy the below and run python3 main.py. The app will create a local file test.txt and write a Unix timestamp to it every second.
2. Stop the app and comment out @app.on_event('startup')
3. Run the app again. The file test.txt is no longer being updated. If you delete this file and restart the app, it will never be created.

import time
from fastapi import FastAPI
from fastapi_utils.tasks import repeat_every
import uvicorn

app = FastAPI()

@app.on_event('startup')
@repeat_every(seconds=1)
def do_something():
    with open('test.txt', 'w') as f:
        f.write(f"{time.time():.0f}")

if __name__ == '__main__':
    uvicorn.run('main:app')

Expected behavior
The function do_something() should be writing to the file every 1 second, even with @app.on_event('startup') commented out.

Environment:

• OS: Windows 10
• FastAPI Utils: 0.2.1
• FastAPI: 0.75.0
• Pydantic: 1.9.0
• Python: 3.10.4

[BUG]/[FEATURE] multiple path's on the same method results in errors

Not quite clear to me if this is a bug or a feature request as I may be doing something out of the ordinary.

Using class based views, my tests return errors using methods (inside cbv) that have multiple path decorators attached to them. When using vanilla FastAPI without cbv the tests pass.

To Reproduce

Credit to @smparekh for the demo below, which I modified slightly to show the issue.

Simple main.py app:

from fastapi import FastAPI, APIRouter

from fastapi_utils.cbv import cbv

router = APIRouter()


fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]


@cbv(router)
class RootHandler:
    @router.get("/items/?")
    @router.get("/items/{item_path:path}")
    @router.get("/database/{item_path:path}")
    def root(self, item_path: str = None, item_query: str = None):
        if item_path:
            return {"item_path": item_path}
        elif item_query:
            return {"item_query": item_query}
        else:
            return fake_items_db


app = FastAPI()
app.include_router(router)

simple test_main.py

from .main import router


from starlette.testclient import TestClient
from .main import fake_items_db

client = TestClient(router)


def test_item_path():
    resp = client.get("items/Bar")
    assert resp.status_code == 200
    assert resp.json() == {"item_path": "Bar"}


def test_item_query():
    resp = client.get("items/?item_query=Bar")
    assert resp.status_code == 200
    assert resp.json() == {"item_query": "Bar"}


def test_list():
    resp = client.get("items/")
    assert resp.status_code == 200
    assert resp.json() == fake_items_db


def test_database():
    resp = client.get("database/")
    assert resp.status_code == 200
    assert resp.json() == fake_items_db

traceback

traceback from the first test (others are the same)

=================================== FAILURES ===================================
________________________________ test_item_path ________________________________

    def test_item_path():
>       resp = client.get("items/Bar")

app/test_main_mult_decorators.py:11: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../../../miniconda3/envs/web-server-eval/lib/python3.8/site-packages/requests/sessions.py:546: in get
    return self.request('GET', url, **kwargs)
../../../miniconda3/envs/web-server-eval/lib/python3.8/site-packages/starlette/testclient.py:413: in request
    return super().request(
../../../miniconda3/envs/web-server-eval/lib/python3.8/site-packages/requests/sessions.py:533: in request
    resp = self.send(prep, **send_kwargs)
../../../miniconda3/envs/web-server-eval/lib/python3.8/site-packages/requests/sessions.py:646: in send
    r = adapter.send(request, **kwargs)
../../../miniconda3/envs/web-server-eval/lib/python3.8/site-packages/starlette/testclient.py:243: in send
    raise exc from None
../../../miniconda3/envs/web-server-eval/lib/python3.8/site-packages/starlette/testclient.py:240: in send
    loop.run_until_complete(self.app(scope, receive, send))
../../../miniconda3/envs/web-server-eval/lib/python3.8/asyncio/base_events.py:612: in run_until_complete
    return future.result()
../../../miniconda3/envs/web-server-eval/lib/python3.8/site-packages/starlette/routing.py:550: in __call__
    await route.handle(scope, receive, send)
../../../miniconda3/envs/web-server-eval/lib/python3.8/site-packages/starlette/routing.py:227: in handle
    await self.app(scope, receive, send)
../../../miniconda3/envs/web-server-eval/lib/python3.8/site-packages/starlette/routing.py:41: in app
    response = await func(request)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

request = <starlette.requests.Request object at 0x118cd3670>

    async def app(request: Request) -> Response:
        try:
            body = None
            if body_field:
                if is_body_form:
                    body = await request.form()
                else:
                    body_bytes = await request.body()
                    if body_bytes:
                        body = await request.json()
        except Exception as e:
            logger.error(f"Error getting request body: {e}")
            raise HTTPException(
                status_code=400, detail="There was an error parsing the body"
            ) from e
        solved_result = await solve_dependencies(
            request=request,
            dependant=dependant,
            body=body,
            dependency_overrides_provider=dependency_overrides_provider,
        )
        values, errors, background_tasks, sub_response, _ = solved_result
        if errors:
>           raise RequestValidationError(errors, body=body)
E           fastapi.exceptions.RequestValidationError: 1 validation error for Request
E           query -> self
E             field required (type=value_error.missing)

../../../miniconda3/envs/web-server-eval/lib/python3.8/site-packages/fastapi/routing.py:145: RequestValidationError

Expected behavior
Ideally tests work as in vanilla FastAPI without cbv.

Environment:

• OS: tested on macOS and Linux

>>> import fastapi_utils
>>> import fastapi
>>> import pydantic.utils
>>> import pytest
>>>
>>> print(fastapi_utils.__version__)
0.2.0
>>> print(fastapi.__version__)
0.52.0
>>> print(pydantic.utils.version_info())
             pydantic version: 1.4
            pydantic compiled: False
                 install path: /Users/bfalk/miniconda3/envs/web-server-eval/lib/python3.8/site-packages/pydantic
               python version: 3.8.1 (default, Jan  8 2020, 16:15:59)  [Clang 4.0.1 (tags/RELEASE_401/final)]
                     platform: macOS-10.14.6-x86_64-i386-64bit
     optional deps. installed: ['typing-extensions']
>>> print(pytest.__version__)
5.3.5

[FEATURE] Run repeated tasks until stopping condition.

I would like to run repeated tasks until a stopping condition, however as I am aware there's no way of doing this with the existing implementation. There is only a provision for maximum repetitions.

[QUESTION] CBV Inheritance

Description

How might I go about writing a CBV that can be inherited from? I want to make a generic class that just needs the class variables set to populate the default CRUD operations that are in the generic base class.

"First" attempt

cbv_router = APIRouter()

class GenericCBV:
    service: Optional[Service] = Depends(None)
    model: Optional[Vertex] = Depends(None)

    @property
    def model_id(self):
        return f'{self.model.vertex_label()}_id'

    @cbv_router.get('/')
    async def get_model_list(self):
        return self.service.get(self.model.vertex_label())

router = APIRouter()
router.include_router(cbv_router)

@cbv(router)
class InvalidationController(GenericCBV):
    service = Depends(Invalidations)
    model = Depends(Invalidation)

I am pretty sure I am either missing something or this isn't possible yet. Thanks!

Add a decorator like @repeat_at?

Hi!
I find myself wanting a decorator like @repeat_at(cron="0 0 13 * * *") to run the task at 1 pm every day, if I where to implement something like that would you consider merging it to this repo?

probably using croniter for the parsing and just getting the numbers of seconds to sleep and just using the same logic as repeat_every

Why is "self" shown as query parameter in "/docs"?

Description

For some reason the self parameter is displaying for one endpoint, when it shouldn't

# ../products/api.py

from fastapi import Depends
from fastapi_utils.cbv import cbv
from fastapi_utils.inferring_router import InferringRouter

from .models import Product
from .connections import DatabaseConnection

from typing import Optional, List


router = InferringRouter()


@cbv(router)
class ProductAPI:

    db: DatabaseConnection = Depends(DatabaseConnection)

    @router.get("/")
    async def products(self, search: Optional[str] = None) -> List[Product]:
        """List all products"""

        return self.db.get_products(search)

    @router.get("/{id}")
    async def products(self, id: int) -> Product:
        """Get product by id"""

        return self.db.get_product(id)

    @router.post("/")
    async def add_product(self, product: Product):
        """Create a new product entry"""

        self.db.add_product(product)

        return {"message": "success"}


# ./main.py

import typing as t

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from products import ProductAPIRouter

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_credentials=True,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# routes - api
app.include_router(ProductAPIRouter, prefix="/api/products")

Docs

simplifiy_operation_ids is sometimes mangling API names

I'm using simplify_operation_ids to generate user friendly client API names. Most of the time this works well, but sometimes the names get mangled. I've created a simplified example below.

from fastapi import FastAPI
from pydantic import BaseModel

from fastapi_utils.openapi import simplify_operation_ids

class Item(BaseModel):
    name: str

app = FastAPI()

#this works
@app.post("/items/")
async def list_items(item1: Item):
    return item1

#this generates a mangled name
@app.post("/items_bad/")
async def list_items_bad(item1: Item, item2: Item):
    return item1

simplify_operation_ids(app)

With the example above, the generated client APIs (after using openapi-generator) look like:
api_instance.list_items(item) api_instance.list_items_bad(body_list_items_bad_items_bad_post)

I would like the 2nd API to look like
api_instance.list_items_bad(item, item)

Dependabot can't resolve your Python dependency files

Dependabot can't resolve your Python dependency files.

As a result, Dependabot couldn't update your dependencies.

The error Dependabot encountered was:

Creating virtualenv fastapi-utils-gwuG-z_Y-py3.8 in /home/dependabot/.cache/pypoetry/virtualenvs
Updating dependencies
Resolving dependencies...

  PackageNotFound

  Package markdown-include (0.6.0) not found.

  at /usr/local/.pyenv/versions/3.8.6/lib/python3.8/site-packages/poetry/repositories/pool.py:144 in package
      140│                     self._packages.append(package)
      141│ 
      142│                     return package
      143│ 
    → 144│         raise PackageNotFound("Package {} ({}) not found.".format(name, version))
      145│ 
      146│     def find_packages(
      147│         self, dependency,
      148│     ):

If you think the above is an error on Dependabot's side please don't hesitate to get in touch - we'll do whatever we can to fix it.

View the update logs.

[BUG] repeat_every runs 4x per minute when specified to repeat every 60sec

Describe the bug
I'm using repeat_every as in

@app.on_event("startup")
@repeat_every(seconds=60)
def scrumbot_alert():
    """
    Sends alert
    """
    now_tz = datetime.datetime.utcnow().astimezone(pytz.timezone(timezone))
    logging.error(now_tz)
    alert()

But when I put a log to track when that function is run, I get 4 logs for every minute.

ERROR:root:2022-04-15 08:20:28.842180-07:00
ERROR:root:2022-04-15 08:20:30.325091-07:00
ERROR:root:2022-04-15 08:20:35.036041-07:00
ERROR:root:2022-04-15 08:20:35.333271-07:00
ERROR:root:2022-04-15 08:21:28.843411-07:00
ERROR:root:2022-04-15 08:21:30.326840-07:00
ERROR:root:2022-04-15 08:21:35.037719-07:00
ERROR:root:2022-04-15 08:21:35.334868-07:00
ERROR:root:2022-04-15 08:22:28.844427-07:00
ERROR:root:2022-04-15 08:22:30.333721-07:00
ERROR:root:2022-04-15 08:22:35.039823-07:00
ERROR:root:2022-04-15 08:22:35.337983-07:00

Expected behavior
Should see only one log per minute

Environment:
OS: Ubuntu 20.04
fastapi utils: 0.2.1
fastapi: 0.60.1
pydantic: 1.4
python: 3.8

[BUG] Unable to resolve multiple handlers for the same HTTP action but with different routes

Describe the bug

I have a BlogCBV where I have a few handler functions, three of which are GETs:

• get_drafts
• get_blogs
• get_blog

get_drafts has a path of blogs/drafts
get_blog has a path of blogs/{blog_id}

However, when I want to reach the endpoint get_drafts by hitting blogs/drafts, get_blog will be reached instead.

Below is the code snippet:

# blog.py

from typing import List

from fastapi import APIRouter, Depends, Path
from fastapi_utils.cbv import cbv
from fastapi_utils.inferring_router import InferringRouter
from sqlalchemy.orm import Session

from app.api.utils.db import get_db
from app.api.utils.security import get_current_active_user
from app.db import crud
from app.db.orm.blog import Blog
from app.db.orm.user import User
from app.dtos.blog import BlogForDetail, BlogForEdit, BlogForList, BlogForNew
from app.dtos.msg import Msg

router = APIRouter()

BLOG_DEFAULT_PAGE_SIZE = 10


router = InferringRouter()


@cbv(router)
class BlogCBV:
    session: Session = Depends(get_db)
    current_user: User = Depends(get_current_active_user)

    @router.get("/drafts")
    def get_drafts(
        self, *, page: int = 1, page_size: int = BLOG_DEFAULT_PAGE_SIZE,
    ) -> List[BlogForList]:
        """
        Retrieve blog drafts for the current user.
        """
        drafts = crud.blog.get_drafts(
            self.session, author=self.current_user, page=page, page_size=page_size
        )
        return [BlogForList.from_orm(draft) for draft in drafts]

    @router.get("/{blog_id}")
    def get_blog(
        self, *, blog_id: int = Path(..., title="The ID of the blog to be returned."),
    ) -> BlogForDetail:
        """
        Retrieve a published blog.
        """
        blog = crud.blog.get(self.session, id=blog_id)
        return BlogForDetail.from_orm(blog)

    @router.get("/",)
    def get_blogs(
        self, *, page: int = 1, page_size: int = BLOG_DEFAULT_PAGE_SIZE,
    ) -> List[BlogForList]:
        """
        Retrieve published blogs by all users.
        """
        blogs = crud.blog.get_multi(self.session, page=page, page_size=page_size)
        return [BlogForList.from_orm(blog) for blog in blogs]

# api.py

from fastapi import APIRouter

from app.api.api_v1.endpoints import blogs, login, users, utils, profile

api_router.include_router(blogs.router, prefix="/blogs", tags=["blogs"])

# main.py

app = FastAPI(title=config.PROJECT_NAME, openapi_url="/api/v1/openapi.json")
app.include_router(api_router, prefix=config.API_V1_STR)

Expected behavior
get_drafts should be correctly resolved and reached.

Environment:

• OS: Linux (Fedora 31)
• FastAPI Utils, FastAPI, and Pydantic versions:

0.1.1
0.49.0
             pydantic version: 1.4
            pydantic compiled: True
                 install path: /usr/local/lib/python3.7/site-packages/pydantic
               python version: 3.7.4 (default, Sep 12 2019, 15:40:15)  [GCC 8.3.0]
                     platform: Linux-5.5.5-200.fc31.x86_64-x86_64-with-debian-10.1
     optional deps. installed: ['email-validator', 'devtools']

• Python version: 3.7.4

Additional context
In my test:

    def test_get_all_my_drafts(self, session, test_user_token_headers):
        batch_create_random_blogs(session, is_published=False, num=3)
        batch_create_random_blogs(session, is_published=True, num=2)
        url = app.url_path_for("get_drafts")

        r = client.get(url, headers=test_user_token_headers)
        drafts = r.json()

        debug(drafts)

        assert r.status_code == 200
        assert len(drafts) == 3

The url can be correctly resolved: url: '/api/v1/blogs/drafts'

[QUESTION] Declare dependecy in cbv without assigning to value

Description
Is there a way to declare a dependency in a cbv without assigning a value to a class member?

I can add dependencies in a path operation decorator like this:

@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
    return [{"item": "Foo"}, {"item": "Bar"}]

Is there a analogous way to do this with a cbv? The only way I'm aware of is to do something like this

@cbv
class MyView:
    _noop = Depends(authenticate_token) # I don't need to access the return value of `authenticate_token`

Additional context
N/A

[FEATURE] Dependency injection doesn't work in @repeat_every functions

Is your feature request related to a problem? Please describe.
There is no way to include dependencies in a @repeat_every function (aka service = Depends(get_service)). This means if you've built dependency functions for use with path operations (@app.get decorated functions), you'll have to resolve those (at possibly multiple levels) by hand.

Describe the solution you'd like
Ideally, we'd just add support for this. However FastAPI doesn't have a way to resolve something from outside a path operation, and solve_dependencies seems to require a Request of some kind to work. There was some discussion of trying to make that accessible to third party libraries here.

Describe alternatives you've considered
Assuming that there isn't a simple way to implement this, it might be good to add a note in the documentation just mentioning that @repeat_every does not support Depends, until that becomes available.

[BUG] repeat_every doesn't close the task on shutdown

Describe the bug
repeat_every creates a task which keeps running after app shutdown, causing python to terminate with open tasks.

How get application state in periodic task?

I need to get app.state to retrieve the Postgres pool driver in my periodic task.
Because task repeating every 5 sec.

How can I do this with the fastapi routing case (when app.state is available only in request object)?

In this example, I'm forced to create gspread and asyncpg connections every 5 seconds this is a terrible decision.

[FEATURE] Adjust task delay duration before first call

Is your feature request related to a problem? Please describe.
For now, repeat_every decorator only allows us to delay the first call exactly 1 interval with wait_first set to True. We can't do things like wait for 1 minute, then execute the task every 1 hour.

Describe the solution you'd like
Make wait_first accept an int argument.

@repeat_every(seconds=3600, wait_first=60)
def foo():
    """Wait for 60s, then execute this once per hour."""

@repeat_every(seconds=3600, wait_first=True)
def bar():
    """Wait for 1 hour, then execute this once per hour."""

[BUG] h11._util.LocalProtocolError: Can't send data when our state is ERROR

Describe the bug
A clear and concise description of what the bug is.

Full Traceback:

2023-02-20 07:51:32 ERROR [235][None] main.unhandled_exception_handler:151  While responding to POST UNDISCLOSED: Can't send data when our state is ERROR
Traceback (most recent call last):
  File "/usr/lib/python3.8/site-packages/starlette/middleware/errors.py", line 162, in __call__
    await self.app(scope, receive, _send)
  File "/usr/lib/python3.8/site-packages/starlette/middleware/base.py", line 109, in __call__
    await response(scope, receive, send)
  File "/usr/lib/python3.8/site-packages/starlette/responses.py", line 277, in __call__
    await wrap(partial(self.listen_for_disconnect, receive))
  File "/usr/lib/python3.8/site-packages/anyio/_backends/_asyncio.py", line 662, in __aexit__
    raise exceptions[0]
  File "/usr/lib/python3.8/site-packages/starlette/responses.py", line 273, in wrap
    await func()
  File "/usr/lib/python3.8/site-packages/starlette/middleware/base.py", line 134, in stream_response
    return await super().stream_response(send)
  File "/usr/lib/python3.8/site-packages/starlette/responses.py", line 255, in stream_response
    await send(
  File "/usr/lib/python3.8/site-packages/starlette/middleware/errors.py", line 159, in _send
    await send(message)
  File "/usr/lib/python3.8/site-packages/uvicorn/protocols/http/h11_impl.py", line 494, in send
    output = self.conn.send(event)
  File "/usr/lib/python3.8/site-packages/h11/_connection.py", line 512, in send
    data_list = self.send_with_data_passthrough(event)
  File "/usr/lib/python3.8/site-packages/h11/_connection.py", line 527, in send_with_data_passthrough
    raise LocalProtocolError("Can't send data when our state is ERROR")
h11._util.LocalProtocolError: Can't send data when our state is ERROR
2023-02-20 07:51:32 INFO  [236][None] timing.emit:132  TIMING: Wall: 3115.3ms | CPU:  348.7ms | UNDISCLOSED

To Reproduce
Enable the timing middleware and use fastapi >= 0.90.0 (0.89.1 is the latest version where the error does not occur). Then call multiple endpoints concurrently and wait. The error does not always happen, but it's not hard to reproduce it with enough ~50-200 concurrent requests.

Expected behavior
The error should not happen.

Environment:

• OS: Alpine Linux (docker alpine:3.12)
• Python 3.8.10

>>> print(fastapi_utils.__version__)
0.2.1
>>> print(fastapi.__version__)
0.92.0
>>> print(pydantic.utils.version_info())
             pydantic version: 1.10.5
            pydantic compiled: True
                 install path: /usr/lib/python3.8/site-packages/pydantic
               python version: 3.8.10 (default, May  6 2021, 06:30:44)  [GCC 9.3.0]
                     platform: Linux-4.19.0-23-amd64-x86_64-with
     optional deps. installed: ['dotenv', 'email-validator', 'typing-extensions']

Additional context

Does not happen for previous versions of FastAPI or with the latest version 0.92.0 and the timing middleware disabled.

https://www.starlette.io/release-notes/
https://github.com/tiangolo/fastapi/releases

Something to look at (but might not be the cause since with starlette 0.23.0 it does already fail):

This can solve nuanced errors when using middlewares. Before Starlette 0.24.0, a new instance of each middleware class would be created when a new middleware was added. That normally was not a problem, unless the middleware class expected to be created only once, with only one instance, that happened in some cases. This upgrade would solve those cases (thanks @adriangb! Starlette PR #2017). Now the middleware class instances are created once, right before the first request (the first time the app is called).
If you depended on that previous behavior, you might need to update your code. As always, make sure your tests pass before merging the upgrade.

Might be related to #266.

CBV Router fails when path is empty

Describe the bug
When the router path is empty string, cbv router fails.

  File "./test.py", line 11, in <module>
    @cbv(router)
  File "/Users/vinod/.pyenv/versions/3.6.10/envs/test/lib/python3.6/site-packages/fastapi_utils/cbv.py", line 26, in decorator
    return _cbv(router, cls)
  File "/Users/vinod/.pyenv/versions/3.6.10/envs/test/lib/python3.6/site-packages/fastapi_utils/cbv.py", line 49, in _cbv
    router.include_router(cbv_router)
  File "/Users/vinod/.pyenv/versions/3.6.10/envs/test/lib/python3.6/site-packages/fastapi/routing.py", line 584, in include_router
    f"Prefix and path cannot be both empty (path operation: {name})"
Exception: Prefix and path cannot be both empty (path operation: get_items)

To Reproduce
Steps to reproduce the behavior:

1. Create a file test.py:

from fastapi import APIRouter
from fastapi import FastAPI
from fastapi_utils.cbv import cbv

main_router = APIRouter()
child_router = APIRouter()


@cbv(child_router)
class MyClass:
    @child_router.get("")
    def get_items(self):
        return {"hello": "world"}


main_router.include_router(child_router, prefix="/items")


app = FastAPI()
app.include_router(main_router)

2. Start the server with uvicorn --reload test:app

Expected behavior
The uvicorn server to start normally. The end point http://localhost:8000/items should return {"hello": "world"}

Environment:

• OS: [e.g. macOS]

• FastAPI Utils, FastAPI, and Pydantic versions

  • fastapi==0.55.1
  • fastapi-utils==0.2.1
  • pydantic==1.4

• Python version: 3.6.10

Additional context
This is because because the cbv helper is including a third router under child router. FastAPI throws an error when both path and prefix are empty.
If I use @child_router.get("/") instead of @child_router.get("") and it is working. FastAPI responds with 307 to redirect the user to right endpoint. Since some http clients are not handling 307 correctly, is there a way to support empty path strings with cbv.

Dependabot couldn't authenticate with https://pypi.python.org/simple/

Dependabot couldn't authenticate with https://pypi.python.org/simple/.

You can provide authentication details in your Dependabot dashboard by clicking into the account menu (in the top right) and selecting 'Config variables'.

View the update logs.

[BUG] Router's prefix is added twice when using CBV

Describe the bug

When I use CBV with a router that has a prefix, the prefix is included twice.

from fastapi import APIRouter
from fastapi_utils.cbv import cbv


router = APIRouter(prefix='/api/v1')


@cbv(router)
class C:
    @router.get('')
    def f(self):
        ...


assert router.routes[-1].path == '/api/v1/api/v1'

>>> print(fastapi_utils.__version__)
0.2.1
>>> print(fastapi.__version__)
0.63.0

This is because the path already has a prefix before CBV removes and re-adds it to a router:

1. @router.get calls router.add_api_route which adds a prefix to the path.
2. cbv calls router.include_router(cbv_router) which again calls router.add_api_route which adds a prefix to the path.

One solution would be to not remove and re-add routes here, and only change the signature, so instead of

    for route in cbv_routes:
        router.routes.remove(route)
        _update_cbv_route_endpoint_signature(cls, route)
        cbv_router.routes.append(route)
    router.include_router(cbv_router)

do

    for route in cbv_routes:
        _update_cbv_route_endpoint_signature(cls, route)

[FEATURE] Ability to pass connect_args to FastAPISessionMaker

I need to be able to pass connect_args to FastAPISessionMaker so I can configure the SQLAlchemy engine to correctly connect to an AWS Aurora cluster (which needs aurora_cluster_arn and secret_arn to be passed as connect_args).

At the moment, I'm not able to do this, because there is no easy way to customise the get_engine function which FastAPISessionMaker uses.

question about race condition

fastapi-utils/fastapi_utils/session.py

Line 33 in af95ff4

def cached_engine(self) -> sa.engine.Engine:

hi , if two thread call the cached_engine , and one stoped in line 39 then yeild , and another thread start running and the engine is still none ,then both these two thread will call get_new_engine ,but there must be an engine will be overwrite by another .

may this can happend ?

how can i specify sqlalchemy engine pool size

how can i specify sqlalchemy engine pool size

[QUESTION] Repeated task: Multiple executions when using multiple workers

Hey there,

when i use repeated task in production with a docker gunicorn/uvicorn image there are multiple instances of the application running, each one with the repeated task.
So for example i want to send notifications periodically, the notification will get send multiple times (number of workers)
My workaround for this would be to run a separate container with only one worker, which is then responsible for the repeated task. Is my usecase out of scope or is there a better solution?

best regards,

[FEATURE] Update schema name for 'UploadFile' routes.

This is a problem that arose on FastAPI#1442. In essence, when creating an upload endpoint like the one detailed here, FastAPI automatically generates the schema names for the UploadFile (or bytes) with no customization setting.

This is modifiable via a simple function that updates the model class name:

def update_upload_schema_name(app: FastAPI, function: Callable, name: str) -> None:
    """
    Updates the Pydantic schema name for a FastAPI function that takes
    in a fastapi.UploadFile = File(...) or bytes = File(...).
    """
    for route in app.routes:
        if route.endpoint == function:
            route.body_field.type_.__name__ = name
            break

The example linked above in the FastAPI endpoint therefore becomes:

from fastapi import FastAPI, File, UploadFile
from typing import Callable

app = FastAPI()

@app.post("/files/")
async def create_file(file: bytes = File(...)):
    return {"file_size": len(file)}

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    return {"filename": file.filename}

def update_upload_schema_name(app: FastAPI, function: Callable, name: str) -> None:
    """
    Updates the Pydantic schema name for a FastAPI function that takes
    in a fastapi.UploadFile = File(...) or bytes = File(...).
    """
    for route in app.routes:
        if route.endpoint == function:
            route.body_field.type_.__name__ = name
            break

update_upload_schema_name(app, create_file, "CreateFileSchema")
update_upload_schema_name(app, create_upload_file, "CreateUploadSchema")

And outputs:

I was wondering if the devs of this repo would find this function useful. Since I have not read through the code of this repo, I'm unsure if there are better ways to add this feature. If someone could guide me to where this could possibly be added,
I would be happy to submit PR. 🤙

[BUG] ensure_future reference should be saved until task is end

fastapi-utils/fastapi_utils/tasks.py

Line 78 in 3ef27a6

ensure_future(loop())

https://docs.python.org/3/library/asyncio-future.html#asyncio.ensure_future

Save a reference to the result of this function, to avoid a task disappearing mid-execution.

Support to SqlAlchemy 2.0

Description
Hi there

Thank you for your effort to maintain this project

Is there any plan to support SqlALchemy 2.0?

I'm trying do install, but got the above error

#0 8.704 ERROR: Cannot install -r requirements.txt (line 12), -r requirements.txt (line 2) and SQLAlchemy==2.0.8 because these package versions have conflicting dependencies. #0 8.705 #0 8.705 The conflict is caused by: #0 8.705 The user requested SQLAlchemy==2.0.8 #0 8.705 fastapi-utils 0.2.1 depends on sqlalchemy<2.0.0 and >=1.3.12 #0 8.705 #0 8.705 To fix this you could try to: #0 8.705 1. loosen the range of package versions you've specified #0 8.705 2. remove package versions to allow pip attempt to solve the dependency conflict

[BUG] Depends not execute when use Request directly

Test Code:

from fastapi import Request, FastAPI, Depends, APIRouter
from fastapi_utils.cbv import cbv



def get_locale(request: Request):
    return request.headers.get("locale", "en_US")

router = APIRouter()

@router.get("/test")
def get_test(locale=Depends(get_locale)):
    return {"locale": locale}


@cbv(router)
class TestRouter:
    locale = Depends(get_locale)

    @router.get("/test1")
    def get_test_cbv(self):
        print(self.locale)
        return {"locale": self.locale}

app =FastAPI()

app.include_router(router, prefix="")

result of test is {"locale": "en_US"}
result of test1 is { "locale": { "dependency": {}, "use_cache": true } }

Depends was not execute as print info:

INFO:     Uvicorn running on http://127.0.0.1:8080 (Press CTRL+C to quit)
Depends(get_locale)
INFO:     127.0.0.1:56462 - "GET /test1 HTTP/1.1" 200 OK
Depends(get_locale)
INFO:     127.0.0.1:56462 - "GET /test1 HTTP/1.1" 200 OK

[FEATURE] Make sqlalchemy an optional dependency

Is your feature request related to a problem? Please describe.

We use fastapi_utils.InferringRouter on a lot of projects to enable more natural use of type annotations within FastAPI. On some of these projects, there is no database usage at all or even a different database type all together, such as MongoDB.

We'd like to avoid the need of installing sqlalchemy (and its dependency greenlet) in these situations, which makes the Docker images larger and is one more package we need to track for updates and audit.

Describe the solution you'd like

Be able to install fastapi-utils in two ways:

$ pip install fastapi-utils
# This depends only on fastapi and pydantic

$ pip install fastapi[sqlalchemy]
# This additionally installs the extra-requires sqlalchemy

Additional context

In Poetry, use the extras keyword to accomplish this: https://python-poetry.org/docs/pyproject/#extras

Setuptools docs on optional dependencies: https://setuptools.readthedocs.io/en/latest/userguide/dependency_management.html#optional-dependencies

[QUESTION] Is there a way to run a task at a specific time?

fastiapi-utils is great and simple to use. I love the repeat_every functionality. However, is there a way to run a task at a specific time?

[QUESTION]Is it possible to handle a polling request using repeated task?

Description
Is it possible to handle a polling request using repeated task?

I need to monitor a resonse in real time use a polling request. Is there any callback argument I can use in repeated task?

[QUESTION] What `Wall` means?

Description

When you add a timing middlware I see this on the LOGs:

INFO:app.engine:TIMING: Wall:  320.7ms | CPU:   12.7ms | app.app.users.router.create_user

It is not clear what Wall means... A lot of routes the Wall and CPU values are the same, but sometimes not. I would appreciate any input why this is the case, and what wall means

Additional context

[BUG] Cant import response

Describe the bug
Cant import response module. Not available in Windows Python3.8

[BUG] CBV interaction with url_for

I've just run into this issue in the wild. It's not a bug as such, but perhaps there's a way to prevent it catching people out like it did me?

Consider this example:

from fastapi import APIRouter, FastAPI, Request
from fastapi_utils.cbv import cbv

router = APIRouter()


@cbv(router)
class Foo:
    @router.get("/foo")
    async def example(self, request: Request):
        return request.url_for("example")


@cbv(router)
class Bar:
    @router.get("/bar")
    async def example(self, request: Request):
        return request.url_for("example")


app = FastAPI()
app.include_router(router)

Both of the url_for("example") calls return "/foo", whereas the expected behaviour might be for the second one to return "/bar" instead.

Without CBVs, IDEs and linters would warn about the function example being redefined, but it obviously doesn't when they're in two different classes. So the problem goes unnoticed until runtime, at which point the 'wrong' path is used.

If we made a change like this to cbv.py...

    for route in cbv_routes:
        router.routes.remove(route)
        _update_cbv_route_endpoint_signature(cls, route)
        route.name = cls.__name__ + '.' + route.name  # <-------------- this line is new
        cbv_router.routes.append(route)
    router.include_router(cbv_router)
    return cls

...then we could do url_for("Foo.example") or url_for("Bar.example").

However:

• It would need to be backwards compatible, although I don't see any immediate problem with inserting the same route twice with two different names
• It wouldn't stop the problem happening if two different modules both had CBVs with the same name, each containing a method with the same name
• It wouldn't play nicely with from x import Y as Z, if this matters - it would still use the name Y.whatever

Do you think this is worth addressing somehow? Thanks!

[BUG] Cannot use get_db as a dependency with repeated task

While using fastapi-utils scheduler, passing Depends(get_db) as a parameter to the function doesn't work.

This is how it works within FastAPI endpoint:

@app.post("/sample_test")
async def sample_test(db: Session = Depends(get_db)):
    return db.query(models.User.height).all()

This is how it doesn't work with FastAPI utils:

@app.on_event("startup")
@repeat_every(seconds=10)
async def sample_test(db: Session = Depends(get_db)):
    return db.query(models.User.height).all()

(note that the functions are the same, only decorators have changed)

There is no error, but the query is not being called (I think the session is not created?). It just gets stuck.

The get_db() function differs a little bit between those two, but under the hood it does the same for both.

Environment:

• OS: Windows 10
• Python 3.9.5 with:
  • FastAPI Utils 0.2.1
  • FastAPI 0.63.0
  • SQLAlchemy 1.4.22
• SQL Server v.18.8

[QUESTION] There's a way to add a custom decorator to a class-based view?

I'm trying to do something like the following:

@cbv(users_router)
@ResponseHandler.class_decorator
class Users:

    controller = UserController()

    @users_router.post("/users", status_code=201)
    async def create_user(self, user_data: Dict, response: Response) -> Dict:
        return self.controller.create_user(**user_data)

    @users_router.get("/users/{user_id}", status_code=302)
    async def get_user(self, user_id: int, response: Response) -> Dict:
        return self.controller.obtain_user(user_id)

    @users_router.get("/users", status_code=302)
    async def get_all_users(self, response: Response) -> Dict:
        return self.controller.obtain_all_users()

The class_decorator decorator adds custom response for each one of the requests, but when I try to execute one of the services, then this error appears:

{"detail":[{"loc":["query","self"],"msg":"field required","type":"value_error.missing"}]}

Typing issue when using class based views [BUG]

Describe the bug
When I tried to use @cbv(router) and added a class attribute with Depends, my pylint started complaining with E1101: Instance of 'Depends' has no '...' member (no-member).

To Reproduce

from fastapi import APIRouter, Path, Depends, Query
from fastapi_utils.cbv import cbv

router = APIRouter()

class ControllerClass:

    def get_sample() -> str:
        return "sample"

def get_controller() -> ControllerClass:
     return ControllerClass()

@cbv(router)
class MyRouter:

    controller: ControllerClass = Depends(get_controller)

    @router.get("/sample")
    async def sample(self) -> str:
        return self.controller.get_sample() # <- this line recognizes self.controller as `Depends` instance, not `ControllerClass`

# > E1101: Instance of 'Depends' has no 'get_sample' member (no-member)

Although, functionally everything works just fine.

Expected behavior
The typing of the class var should be inferred as ControllerClass, not Depends.

Environment:

• OS: MacOS 19.5.0 Darwin Kernel Version
• FastAPI Utils, FastAPI, and Pydantic versions:

>>> print(fastapi_utils.__version__)
0.2.1
>>> print(fastapi.__version__)
0.55.1
>>> print(pydantic.utils.version_info())
             pydantic version: 1.5.1
            pydantic compiled: True
               python version: 3.7.7 (default, May  6 2020, 04:59:01)  [Clang 4.0.1 (tags/RELEASE_401/final)]
                     platform: Darwin-19.5.0-x86_64-i386-64bit
     optional deps. installed: ['typing-extensions']

• Python version, get it with: Python 3.7.7

[QUESTION] subrouting with cbv

Description

How can I use class based subrouting?
Please show me example for this, I have big app and don't found simple example on the site

[QUESTION]

Trying to create generic CRUD class for all endpoints

I am fairly new to FastAPI(migrating from Django) and I am trying to create a generic CRUD operations class that I can inherit and use across my CBV endpoints.
Something like this :

class AbstractCrud
      model: Base = NotImplemented
      session: Session = NotImplemented
  
      def get_items(self, limit,  **filters):
          """ Read operation """
  
      def get_item(self, pk: int):
  
  
      def create_item(self, obj: BaseModel):
          """ Create operation """
  
      def update_item(self, pk: int, **params):
          """ Update operation"""


      def delete_item(self, pk: int):
        """ Delete operation """


router = InferringRouter()

@cbv(router)
class UserAPI(AbstractCrud):
    router.tags = ["User"]
    router.prefix = "/users"
   
    model = User
    session: Session = Depends(get_db)

   # my endpoints
   #e.g. @router.get(...)


@cbv(router)
class PostAPI(AbstractCrud):
    router.tags = ["Post"]
    router.prefix = "/posts"

    model = Post
    session: Session = Depends(get_db)

    # my endpoints
    #e.g. @router.get(...)

I get the following error if I try to do the above:
fastapi.exceptions.FastAPIError: Invalid args for response field! Hint: check that <class 'sqlalchemy.orm.decl_api.Base'> is a valid pydantic field type
For now, I am able to achieve this as follows:

class AbstractCrud
      model: Base = NotImplemented
      session: Session = NotImplemented
  
      def get_items(self, limit,  **filters):
          """ Read operation """
  
      def get_item(self, pk: int):
  
  
      def create_item(self, obj: BaseModel):
          """ Create operation """
  
      def update_item(self, pk: int, **params):
          """ Update operation"""


      def delete_item(self, pk: int):
        """ Delete operation """
        

class UserCrud(AbstractCrud):

    def __init__(self, session: Session):
        self.session = session
        self.model = User


class PostCrud(AbstractCrud):

    def __init__(self, session: Session):
        self.session = session
        self.model = Post


router = InferringRouter()

@cbv(router)
class UserAPI:
    router.tags = ["User"]
    router.prefix = "/users"
   
    def __init__(self, session=Depends(get_db)):
        self.session = session
        self.crud = UserCrud(self.session)

   # my endpoints
   #e.g. @router.get(...)


@cbv(router)
class PostAPI:
    router.tags = ["Post"]
    router.prefix = "/posts"

    def __init__(self, session=Depends(get_db)):
        self.session = session
        self.crud = PostCrud(self.session)

    # my endpoints
    #e.g. @router.get(...)

Although this is working fine for me now, I can't help but think if there is a better(or correct) way to do this.

Also, Is my use of a single router variable across multiple classes correct?

Error handling that has access to cbv

I would like to make a class based view that can define its own error handling. I know that I can define global error handlers using @app.exception_handler(MyException). But I would love to be able to write an error handler that some how has access to the cbv instance so I can access state that was built up while processing the request. I think this would lead to cleaner code than wrapping everything endpoint in a try/except

I'm imagining something like this:

@cbv(router)
Class MyView:
    @router.get('/hello')
    def hello(self):
        return {'msg': 'hello world'}

    def handle_errors(self):
        """errors happening in this class are routed to here"""

If I was using Flask, there would be a way to do this with their class based view system

I might do something like this

from flask.views import View

class BaseRoute(View):
    def __init__(self, *args, **kwargs):
        """
        do setup here
        """

    def dispatch_request(self, *args, **kwargs):
        try:
            # self.work is implemented by the sun class
            return self.work(*args, **kwargs)
        except MyError as e:
            """
            handle my error
            """
        finally:
            """
            clean up
            """

class GetUsers(BaseRoute):
    def work(self, *args, **kwargs):
        """
        do the actual work of the endpoint
        """

But of course since FastAPI aims to be smarter about types, this is less trivial (after a few hours of looking around I didn't see an obvious way).

Any ideas on how to go about something like this? I'm flexible with the approach to solving it, at the end of the day I want access to self when handling errors without having to do a try/except in every endpoint

[BUG] @repeat_every will have no stack trace in terminal

Describe the bug
No error will show

To Reproduce

from fastapi import FastAPI

from fastapi_utils.tasks import repeat_every


app = FastAPI()

items = {}


@app.on_event("startup")
@repeat_every(seconds=60)
async def startup_event():
    raise Exception
    items["foo"] = {"name": "Fighters"}
    items["bar"] = {"name": "Tenders"}


@app.get("/items/{item_id}")
async def read_items(item_id: str):
    return items[item_id]

uvicorn app.main:app --reload, the raise Exception should produce a stack trace in terminal, but none.

Environment:

• Windows

Python 3.7.7 (tags/v3.7.7:d7c567b08f, Mar 10 2020, 10:41:24) [MSC v.1900 64 bit (AMD64)] on win32
Type "help", "copyright", "credits" or "license" for more information.
>>> import fastapi_utils
>>> import fastapi
>>> import pydantic.utils
>>> print(fastapi_utils.__version__)
0.2.1
>>> print(fastapi.__version__)
0.58.0
>>> print(pydantic.utils.version_info())
             pydantic version: 1.5.1
            pydantic compiled: True
                 install path: D:\git\testFastapiutils\venv\Lib\site-packages\pydantic
               python version: 3.7.7 (tags/v3.7.7:d7c567b08f, Mar 10 2020, 10:41:24) [MSC v.1900 64 bit (AMD64)]
                     platform: Windows-10-10.0.18362-SP0
     optional deps. installed: []

[QUESTION] GUID Type: Use UUID.hex instead of UUID.int?

For the GUID Type, the code from SQLAlchemy says that for non-PostgreSQL databases, it's "storing as stringified hex values"… but as far as I understand this code, it's actually storing as 128-bit integer because UUID.int is used.

It should use UUID.hex.

u = uuid4()  # UUID('33be8037-6c92-41d3-a5f7-018fa1175354')
u.int        # 68779764727731727554280205683269260116
u.hex        # '33be80376c9241d3a5f7018fa1175354'

[BUG] cbv and path params create strange behaviour

While using the @cbv i encountered the problem that i couldn't use the endpoint without path params on a similar path.
It somehow messes with the validator, i think?

from fastapi import APIRouter, FastAPI
from fastapi_utils.cbv import cbv
from pydantic.types import UUID4
from starlette.testclient import TestClient

router = APIRouter()


@cbv(router)
class CBV:

    @router.post("/test")
    def test_post(self):
        return ''


    @router.get("/test/{uuid}")
    def test_post(self, uuid: UUID4):
        return uuid


app = FastAPI()
app.include_router(router)
client = TestClient(app)

print(client.post("/test").json())

output:

{
  "detail":[
    {
      "loc":[
        "query",
        "self"
      ],
      "msg":"field required",
      "type":"value_error.missing"
    }
  ]
}

Expected behavior
both endpoints should be just available

Environment:

fastapi_utils_version: 0.2.1
fastapi_version: 0.85.2
pydantic version: 1.10.2
pydantic compiled: True
install path: ...\fastapi_test\venv\Lib\site-packages\pydantic
python version: 3.10.8 (tags/v3.10.8:aaaf517, Oct 11 2022, 16:50:30) [MSC v.1933 64 bit (AMD64)]
platform: Windows-10-10.0.22621-SP0
optional deps. installed: ['dotenv', 'typing-extensions']

[FEATURE] FastAPISessionMaker documentation

Hi is there an example somewhere on how to use the SessionMaker?
Thank you!

[BUG] InferringRouter, typing and response without body

Python 3.7
Fastapi and fastapi-utils the latest
Ubuntu 18.04

It seems that InferringRouter create response_model in case when this model has not expected

Example, this code is ok:

from typing import NoReturn, Optional

from fastapi import APIRouter

router = APIRouter()

@router.delete('/{item_id}', status_code=status.HTTP_204_NO_CONTENT)
def delete_item(*, item_id: int) -> Optional[NoReturn]:
    resource.get_or_404(obj_id=item_id)
    resource.remove(obj_id=item_id)

let's use InferringRouter instead APIRouter

class InferringRouter(APIRouter):
    def add_api_route(self, path: str, endpoint: Callable[..., Any], **kwargs: Any) -> None:
        if kwargs.get("response_model") is None:
            kwargs["response_model"] = get_type_hints(endpoint).get("return")
        return super().add_api_route(path, endpoint, **kwargs)

from fastapi_utils.inferring_router import InferringRouter

router = InferringRouter()

@router.delete('/{item_id}', status_code=status.HTTP_204_NO_CONTENT)
def delete_item(*, item_id: int) -> Optional[NoReturn]:
    ...

exception!

AssertionError: Status code 204 must not have a response body

class APIRoute(routing.Route):
    def __init__(
        ...
        response_model: Type[Any] = None,
        ...
    ) -> None:
        self.response_model = response_model
        if self.response_model:
            assert (
                status_code not in STATUS_CODES_WITH_NO_BODY
            ), f"Status code {status_code} must not have a response body"

all of this examples failed:

def delete_item(*, item_id: int) -> Optional[NoReturn]:

def delete_item(*, item_id: int) -> None:

but example without typing works:

def delete_item(*, item_id: int):

is it possible to use InferringRouter for STATUS_CODES_WITH_NO_BODY with typing result of method?

[BUG] InferringRouter throws exception with FileResponse return type

Describe the bug

Setting the return type to -> FileResponse generates an exception:

Traceback (most recent call last):
  File "/venv/lib/python3.9/site-packages/fastapi/utils.py", line 65, in create_response_field
    return response_field(field_info=field_info)
  File "pydantic/fields.py", line 342, in pydantic.fields.ModelField.__init__
  File "pydantic/fields.py", line 456, in pydantic.fields.ModelField.prepare
  File "pydantic/fields.py", line 670, in pydantic.fields.ModelField.populate_validators
  File "pydantic/validators.py", line 715, in find_validators
RuntimeError: no validator found for <class 'starlette.responses.FileResponse'>, see `arbitrary_types_allowed` in Config

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "runpy.py", line 197, in _run_module_as_main
    return _run_code(code, main_globals, None,
  File "runpy.py", line 87, in _run_code
    exec(code, run_globals)
  File "/venv/lib/python3.9/site-packages/uvicorn/__main__.py", line 4, in <module>
    uvicorn.main()
  File "/venv/lib/python3.9/site-packages/click/core.py", line 1137, in __call__
    return self.main(*args, **kwargs)
  File "/venv/lib/python3.9/site-packages/click/core.py", line 1062, in main
    rv = self.invoke(ctx)
  File "/venv/lib/python3.9/site-packages/click/core.py", line 1404, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/venv/lib/python3.9/site-packages/click/core.py", line 763, in invoke
    return __callback(*args, **kwargs)
  File "/venv/lib/python3.9/site-packages/uvicorn/main.py", line 371, in main
    run(app, **kwargs)
  File "/venv/lib/python3.9/site-packages/uvicorn/main.py", line 393, in run
    server.run()
  File "/venv/lib/python3.9/site-packages/uvicorn/server.py", line 50, in run
    loop.run_until_complete(self.serve(sockets=sockets))
  File "asyncio/base_events.py", line 642, in run_until_complete
    return future.result()
  File "/venv/lib/python3.9/site-packages/uvicorn/server.py", line 57, in serve
    config.load()
  File "/venv/lib/python3.9/site-packages/uvicorn/config.py", line 318, in load
    self.loaded_app = import_from_string(self.app)
  File "/venv/lib/python3.9/site-packages/uvicorn/importer.py", line 22, in import_from_string
    module = importlib.import_module(module_str)
  File "importlib/__init__.py", line 127, in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
  File "<frozen importlib._bootstrap>", line 1030, in _gcd_import
  File "<frozen importlib._bootstrap>", line 1007, in _find_and_load
  File "<frozen importlib._bootstrap>", line 986, in _find_and_load_unlocked
  File "<frozen importlib._bootstrap>", line 680, in _load_unlocked
  File "<frozen importlib._bootstrap_external>", line 855, in exec_module
  File "<frozen importlib._bootstrap>", line 228, in _call_with_frames_removed
  File "/./main.py", line 14, in <module>
    async def root() -> FileResponse:
  File "/venv/lib/python3.9/site-packages/fastapi/routing.py", line 551, in decorator
    self.add_api_route(
  File "/venv/lib/python3.9/site-packages/fastapi_utils/inferring_router.py", line 16, in add_api_route
    return super().add_api_route(path, endpoint, **kwargs)
  File "/venv/lib/python3.9/site-packages/fastapi/routing.py", line 496, in add_api_route
    route = route_class(
  File "/venv/lib/python3.9/site-packages/fastapi/routing.py", line 324, in __init__
    self.response_field = create_response_field(
  File "/venv/lib/python3.9/site-packages/fastapi/utils.py", line 67, in create_response_field
    raise fastapi.exceptions.FastAPIError(
fastapi.exceptions.FastAPIError: Invalid args for response field! Hint: check that <class 'starlette.responses.FileResponse'> is a valid pydantic field type

To Reproduce

from fastapi import FastAPI
from fastapi.responses import FileResponse
from fastapi_utils.inferring_router import InferringRouter

app = FastAPI()
router = InferringRouter()


@router.get("/")
async def root() -> FileResponse:
    return FileResponse("file.txt")

app.include_router(router)

Launch with:

$ python -m uvicorn main:app

Expected behavior

A return type of -> FileReponse should not throw an exception. Allowing this allows static type checkers such as mypy to validate that all code paths in the path function return the correct type.

Environment:

• OS: macOS
• FastAPI Utils, FastAPI, and Pydantic versions:

pip list                  
Package           Version
----------------- --------
aiofiles          0.7.0
asgiref           3.3.4
click             8.0.1
fastapi           0.65.1
fastapi-utils     0.2.1
greenlet          1.1.0
h11               0.12.0
pip               21.1.2
pydantic          1.8.2
setuptools        57.0.0
SQLAlchemy        1.4.17
starlette         0.14.2
typing-extensions 3.10.0.0
uvicorn           0.14.0

0.2.1
0.65.1
             pydantic version: 1.8.2
            pydantic compiled: True
                 install path: /Users/hagenjt1/PycharmProjects/fastapi-test/venv/lib/python3.9/site-packages/pydantic
               python version: 3.9.5 (default, May  4 2021, 03:33:11)  [Clang 12.0.0 (clang-1200.0.32.29)]
                     platform: macOS-10.15.7-x86_64-i386-64bit
     optional deps. installed: ['typing-extensions']

• Python version: 3.9.5

Additional context

Directly setting the response_class on vanilla FastAPI does not throw an exception:

# This launches fine:
from fastapi import FastAPI
from fastapi.responses import FileResponse

app = FastAPI()


@app.get("/", response_class=FileResponse)
async def root():
    return FileResponse("file.txt")

[QUESTION]

Description
I need to asynchronously get some config from my config server using repeated task.

I declare a function to asynchronously get config decorated with the @app.on_event("startup") and @repeat_every.
like this:
@app.on_event("startup")
@repeat_every(seconds=60)
async def listener_task() -> None:

And I use the config in another function decorated with the @app.on_event("startup").
like this:
@app.on_event("startup")
async def init_orm() -> None:

How can I ensure the first time I get the config before using the config.

Additional context
Add any other context or screenshots about the feature request here.

[BUG] Does not import on Windows

Describe the bug
Fails completely on Windows because resource is not available on Windows.

\lib\site-packages\fastapi_utils\timing.py", line 10, in <module> import resource
ModuleNotFoundError: No module named 'resource'

To Reproduce
Steps to reproduce the behavior:

1. pip install fastapi-utils
   2.from fastapi_utils.timing import add_timing_middleware

Expected behavior
The import works.

Environment:

• OS: Windows 10
• Python version: 3.7

Additional context
https://stackoverflow.com/questions/37710848/importerror-no-module-named-resource