Thank you again for this useful repo.
I have a question regarding your tip 23. I wonder what the differences between using def function and run_in_threadpool in async function. They works the same way to my knowledge. My use case is that I'm using google_storage_python to download file from GCS but it's a blocking I/O library

in this repo the highest level of app is src/

fast-api-project/
└── src/
    ├── domain1
    └── domain2

and then i see the Dispatch repo that mention on the description
highest level of app is src/distpatch/

dispatch/
└── src/
    └── dispatch/
        ├── domain1
        └── domain2

dispatch folder inside src is called domain to?
what should i choose?

is there anyway to call async function in pydantic field validator?

like this example?

i found this on pydantic repo issues

do you have any solution for this?

from pydantic import BaseModel, validator


async def is_admin(id) -> bool:
    # some query to db
    return True

class IsAdminSchema(BaseModel):
    user_id: int
        
    @field_validator('user_id')
    @classmethod
    async def user_id_validator(cls, v: int) -> int:
        if await is_admin(value):
            return v
        raise Forbidden()

I'm fairly new to both Python and FastAPI, so I might be missing an obvious solution here. I joined a team that's already working with said stack and they've set up a dev Docker image that runs the app through uvicorn. I see a similar setup in your repo.

Coming from Java, I'd prefer to run and debug the application without the need to run it inside a container (and also setting up remote debugging looks like a pain). Why is there a preference to run it in a container? I tried following FastAPI's suggested setup, but I quickly ran into some module import issues (probably due to my lack of deep Python understanding).

Any guidance on the reason for the design choice and possible bast practices would be welcome :)

@zhanymkanov thanks for the write up. It’s great to have some benchmarks on professional implementations, this is awesome and one of the most valuables repositories, just a lot of production-ready and architecture tips, great stuff, thanks a lot for sharing this!

To our (very positive) surprise, this is very similar to what we are doing in our side. I though it was worth sharing our experiences and the choices we've made along the way good and bad.

⚠️ This ended up being a lot longer than what I expected, my apologies.

Project Structure

This is very similar to what we are doing. The functional way of splitting things doesn’t really work except for really small projects, so we also have a “module” based approach. Our application looks something like:

ourproject-backend
├── alembic/
├── app
│   ├── auth
│   │   ├── routes.py
│   │   ├── schemas.py  # pydantic models
│   │   ├── models.py  # db models
│   │   ├── permissions.py # our decorator
│   │   ├── exceptions.py
│   │   ├── service.py
│   │   └── utils.py
│   ├── core
│   │   ├── routes.py
│   │   ├── services.py
│   │   ├── ....
│   ├── users
│   │   ├── routes.py
│   │   ├── services.py
│   │   ├── ....
│   ├── tenants
│   │   ├── routes.py
│   │   ├── services.py
│   │   ├── ....
│   ├── extensions
│   │   ├── logs.py # JSON Logger etc
│   │   ├── middleware.py # correlation ID & request tracker
│   │   ├── ....
│   ├── services
│   │   ├── mailer.py # a client to SES
│   │   ├── filesystem.py #  a wrapper over S3
│   │   ├── ....
│   ├── db
│   │   ├── mixin.py
│   │   ├── base.py
│   │   ├── engine.py
│   │   ├── ....
│   ├── utils
│   │   ├── schemas.py
│   │   ├── helpers.py
│   │   ├── ....
│   ├── modules
│   │   ├── module_a
│   │   │   ├── models.py
│   │   │   ├── routes.py
│   │   │   ├── schemas.py
│   │   │   ├── ....
│   │   ├── module_b
│   │   │   ├── models.py
│   │   │   ├── routes.py
│   │   │   ├── schemas.py
│   │   │   ├── ....
│   ├── config.py # where the Dynaconf singleton lives
│   ├── exceptions.py
│   ├── routes.py # registration of all system routes
│   ├── hub.py # our event hub
│   └── main.py
├── tests/
│   ├── users
│   ├── tenants
│   └── module_a
├── .env
├── .secrets.toml
├── .gitignore
├── settings.toml
├── mypy.ini
└── alembic.ini

A few comments:

• We use a sort of “mixed” structure in the sense that some global/generic modules (like Users/Tenants/Auth) have all the same structure and are in the top level, but the application specific business logic is in the modules module. We have been using this structure for the past couple of years and have been pretty happy with the separation of concerns it brings. We even reuse the same blueprint for different projects, we mostly just change the modules which is great.
• Having a specific db module on the top level has helped a lot giving us flexibility to have more robust Mixin classes, better engine configuration and some other goodies.
• We also are really happy with having a core module on the top level. This gives us flexibility to do things like a specific mock service, a taskStatus route or more generic resources.
• We really like how predictable this is and much boilerplate code we can just copy around from module to module. We have dramatically speed up our development process of new modules with this. This also helped a lot new devs to understand the codebase logic.

Permissions & Auth

Although the “recommended” way of doing authentication in FastAPI would be the dependency injection, we have chosen to use a class-based decorator to control access on the route level.
So our routes look something like:

@route.get('/me')
@access_control(Resources.users_view_self) # this is a enum
def myroute(self):
...

@route.get('/superuser_only')
@access_control(superuser=True)
def myroute(self):
...


@route.get('/open')
@access_control(open=True)
def myroute(self):
...

And our access_control class looks like:

class access_control:  # pylint: disable=invalid-name
    MASTER_USER_ID = 0

    def __init__(
        cls,
        module: Optional[AppModules] = None,
        resource: Optional[AppActions] = None,
        superuser: bool = False,
        open: bool = False,
    ) -> None:
        cls.module = module
        cls.resource = resource
        cls.superuser = superuser
        cls.open: bool = open
        cls.tenant_id: Optional[int] = None
        cls.object_id: Optional[int] = None
        cls.current_user: Optional[UserResponse] = None
        cls.request: Optional[Request] = None
        cls.headers: Optional[dict[Any, Any]] = None
        cls.auth_header: Optional[str] = None
        cls.token: Optional[str] = None

    def __call__(cls, function) -> Callable[..., Any]:
        @functools.wraps(function)
        async def decorated(*args, **kwargs):
            t0 = time.time()
            try:
                await cls.parse_request(**kwargs)
                is_allowed = await cls.verify_request(*args, **kwargs)
                if not is_allowed:
                    raise HTTPException(403, "Not allowed.")
                return await function(*args, **kwargs)
            except exc.NotAllowed as error:
                raise HTTPException(403, str(error)) from error

        return decorated

    async def parse_request(cls, **kwargs) -> None:
        """Get the current user from the request"""
        dependencies = kwargs.get("self", kwargs.get("base_args"))
        base_args: Optional[RequestArgs] = getattr(dependencies, "base_args", None)
        if not base_args:
            return
        cls.tenant_id = base_args.tenant_id
        cls.current_user = base_args.current_user
        return None

    async def verify_request(cls, **kwargs) -> None:
        """Actually check for permission based on route, user, tenant etc"""
        ...

A few benefits we encountered, and few drawbacks:

• This is great to accept multiple parameters like module or action or superuser=True and things like that.
• The permission controller (the access_control class itself) is fairly easy to work on, being very powerful at the same time, since it has the *args and **kwargs from the request, and the full context (current user, path, tenant, etc), so all sort of checks can be used. As we increase the granularity over access control we have been considering implementing a permissions decorator for each module, so we can have more specific control over a given resource. But WIP still.

Class-based Services

Our service module service.py started to get big and a mess of functions, so we started having a few class based services, which have been working very well. Something like TenantService , UserService. This almost looks like a repository for simple modules (in some cases we even spiltd the service into service and repository (for more complex business logic). Now each service module has anything from 1 to 10 service classes, this greatly improved our organization and readability.

Class-based views

Earlier this year we refactor all of our routes to use a class based view that is included in the fastapi-utils package and this is made our code a lot cleaner. The main benefit for us, is that the basic authentication process (reading the token and the X-Tenant-ID for the header) is done in one place only, se we don’t have to repeat the dependencies.
What we’ve done is, we have a custom commons_deps function, and at the beginning of each route class we do something like:

@cbv(router)
class MyModuleRouter:
    commons = Depends(commons_deps)
    service = MyModuleService()		

    @route.get('/me')
    @access_control(Resources.users_view_self)
    def myroute(self):
         # And now here we can access the common deps & the service
         current_user = self.commons.current_user
         tenant_id = self.commons.tenant_id
         response = self.service.get_module_resource(tenant_id)

We have been experimenting with something slightly different nowadays, which is having the service being instantiated with the tenant_id and current_user in a dependency injection, so that our service starts up a bit more complete.

Task Queues

We are long time Celery users, but celery is overwhelming and fairly difficult to reason about when you get to the internals and specifics. We just switched to RQ and couldn’t be happier with a few caveats. The logic is amazing (the Queue , Job objets are really intuitive and easy to work with, as are dependency chains with depends_on. The thing is that there’s an issue with async functions. They work if you use the worker, but won’t work if you run in the same process, which is kind of a pain when debugging. We haven’t experimented with starlette’s. Background jobs as we always valued having a centralized dashboard for tasks and an easy way to get a task status for example. As we deploy most of our applications in Kubernetes, being able to scale the workers easily and indefinitely is awesome and we are really glad with it. I have been experimenting with a few different snippets to try to open a PR and make RQ compatible in every scenario.

The fancy architecture

In same cases (actually projects) we slightly changed our module architecture to account for a proper business oriented Model object.

...
│   ├── modules
│   │   ├── module_a
│   │   │   ├── routes.py
│   │   │   ├── services.py
│   │   │   ├── orm.py # the sqlalchemy classes
│   │   │   ├── models.py # "pure" modules (are also pydantic)
│   │   │   ├── schemas.py # the pydantic API schemas
│   │   │   ├── adapters.py
│   │   │   ├── builders.py
│   │   │   ├── interfaces.py
│   │   │   ├── repository.py

For fancier implementations this worked very well, although is a lot more complex to start with. This gives us a proper EntityModel and great separation of concerns, but it gets a lot more verbose really quick, so we found it was only worth it for very complex projects, but it’s also a possibility.

Custom Response Serializers & BaseSchema

We found that the response_class in FastAPI also serializes the data in Pydantic, so it’s not purely for documentation. You can, however, overwrite the default response behavior by making a custom response class, which we did going a bit of performance (anywhere from 50-100ms) and flexibility. So we have something like:

# utils/schemas.py

class JSONResponse(Response):
    media_type = "application/json"

    def __init__(
        self,
        content: typing.Any = None,
        status_code: int = 200,
        headers: t.Optional[t.Mapping[str, str]] = None,
        media_type: t.Optional[str] = None,
        background: t.Optional[BackgroundTasks] = None,
    ) -> None:
        self.status_code = status_code
        if media_type is not None:
            self.media_type = media_type
        self.background = background
        self.body = self.render(content)
        self.init_headers(headers)

    def render(self, content: BaseSchema | list[BaseSchema] | Any):
       # This is not 100% battle proof, but as our services are controlled (only return Pydantic modules) works fine
        if isinstance(content, BaseSchema):
            return content.json().encode("utf-8")
        if isinstance(content, list):
            if isinstance(content[0], BaseSchema):
                def uuid_decoder(obj):
                    if isinstance(obj, UUID):
                        return str(obj)
                return orjson.dumps([item.dict() for item in content], default=uuid_decoder)

And then we use the response directly like:

@cbv(router)
class MyModuleRouter:
    commons = Depends(commons_deps)
    service = MyModuleService()		

    @route.get('/me', response_class=[...])
    @access_control(Users.view_self) # this is a enum
    def myroute(self):
        # And now here we can access the commons
      	current_user = self.commons.current_user
      	tenant_id = self.commons.tenant_id
        response = self.service.get_module_resource(tenant_id)
	return JSONResponse(response, 200)

This gave us a cleaner router since we can use the status code on the response itself, which was more intuitive for use, gained a bit of performance with the orjson encoder and we just like it better. The (big) downside is that we face the risk of having documentation/API inconsistencies, in our case it happened once or twice, but we think it’s still worth it.

Just as you guys we also have a BaseSchema base for all Pydantic schemas we use that have a couple of configurations like orm_mode enum etc.

Using a DefaultResponse class

In several occasions the response is kind of generic, so we use a lot of a schema called DefaultResponse:

class DefaultResponse(BaseSchema):
    status: bool
    msg: str
    details: Optional[dict[Any, Any]] = {}

This is a kind of standardized way of communicating with our client (we have a React frontend) so the front devs always know what to look for when getting a DefaultResponse.

Configuration

Although Pydantic is nice for configuration as well, we couldn’t be happier using the amazing @dynaconf lib, developed and maintained by @BrunoRocha. This was a game changer in our settings management.

All of our settings/secrets went to .toml files and a few things happened:
- Only one file for multiple environments using toml headers
- Only one place to manage keys (in Flask we were used of having multiple configuration classes which were a pain to maintain)
- a singleton with global access our settings.py file has ~10 lines:

#app/config.py

from dynaconf import Dynaconf

settings = Dynaconf(
    settings_files=[".settings.toml", ".secrets.toml"],
    envvar_prefix="MYAPP",
    env_switcher="MYAPP_APP_ENV",
    load_dotenv=True,
    environments=True,
)

And now everywhere we can just

from app.config import settings

myvar = settings['MYVAR']
myvar_a = settings.MYVAR_A

And don’t need to change anything when deploying to K8S since we already inject everything with env vars (config). Can’t recommend it more. We still have to experiment with the Vault integration, which is the next step.

The Message Hub

This helped a lot while we were trying to further decouple our services.
The hub is a centralized hub to share message between modules, something like:

class MessageHub:
    """Message hub for events"""

    handlers = {
        module_a.ResourceCreated: [
            module_b.handle_resource_created,
            module_c.handle_resource_created,
        ],
        module_d.ResourceDeleted: [
            module_b.handle_resource_deleted,
            module_c.handle_resource_deleted,
        ],
    }  # type: dict[Type["Event"], list[Callable[..., Any]]]

    @classmethod
    async def track(cls, event: ApplicationEvent):
        """Tracks the Application activity.
        Receives the application event that will be used by the AuditService.

        Args:
            event (ApplicationEvent): The ApplicationEvent
        """
        await AuditService.save(event)

    @classmethod
    async def handle(cls, event: Event):
        """
        Handles an arbitrary event.
        It will receive the event, and get the handlers that should handle
        the event. The order on which the handlers will execute the event may vary.
        If the event is sent to the worker, the handlers are async, meaning they can run at the same time.
        If the event is synchronous, than each handlers will handle the event sequentially.

        Args:
            event (Event): The Event.
        """
        if type(event) not in cls.handlers:
            logger.info("No handlers for event: %s", event.__class__.__name__)
            return

        # Call listeners functions
        for fn in cls.handlers[type(event)]:
            if event.is_async:
                worker.enqueue(fn, event)
                return

            await fn(event)

And in most modules we have handlers.py module that will have a few functions that handle events. The services themselves usually dispatch events, like hub.MessageHub.handle(event_created_by_the_service), and we also use it to track application activity, normally called by the route hub.MessageHub.track(application_activity_schema)

Types & Docs

100% of arguments are typed and 100% of methods / functions have docstrings. I honestly can't live without anymore. Now wondering if could just compile the whole code to C and make it fly? Nuitka, MyPyC maybe? TBC...

Now the bad part, and our (really) bad practices

Local Session Management

For a couple of reasons we didn’t implement the request-coupled session management (inject the session through FastAPI’s dependency injection system) and we ended up having a lot of services that handle the session locally, which is not cool and not recommended by SQLAlchemy. Think of:

class ModuleService:
    ...
    async def module_method(self, ...):
       # Terrible
        async with async_session() as session:
	    ...
	return something

Managing the session lifecycle itself is fairly ok and it works for really simple services, but what we found is that for more complex services methods that call on another you end up nesting sessions which is terrible. Imagine calling other_method from module_method that also has the same session lifecycle management, now you just opened a session within another session. Jus terrible. We are gradually moving to better session management, but we are still trying to find better ways of handling it.

Little use of the Dependency Injection

In your write up a lot of great example of how to properly use and leverage the power of dependency injection, we don’t use much of those, and we definitely should.

Lack of Context in Services

Sometimes we found ourselves having a Service class that didn’t even have a initializer and was purely for organization, this is fine, but we are missing a lot of benefits of having some context in the service (example: tenant_id and session) which would save was from having the tenant_id being passed to every single method in a service class. So there’s definitely a lot to improve here.

There's obviously a lot to improve and a whole lot more of bad things that I probably forgot to mention, but again, I though it was worth sharing the experience. And to finish our Dockerfile, which is also pretty simple (using poetry and leveraging it's dev-dependencies logic something that was mentioned here as well #1 :

FROM python:3.10-slim
WORKDIR /app

COPY pyproject.toml .
COPY poetry.lock* .

RUN apt-get update -y && \
    apt-get install gcc -y && \
    apt-get install libpq-dev -y && \
    python -m venv .venv && \
    .venv/bin/pip install poetry && \
    .venv/bin/poetry export -f requirements.txt --output requirements.txt --no-dev --without-hashes && \
    .venv/bin/pip install -r requirements.txt && \
    apt-get remove gcc -y && \
    apt autoremove -y

ADD . /app
EXPOSE 8000
CMD [".venv/bin/uvicorn", "app.asgi:app", "--host", "0.0.0.0"]

async def valid_post_id(post_id: UUID4) -> Mapping:
    post = await service.get_by_id(post_id)
    if not post:
        raise PostNotFound()

    return post

How should service be written, my question is how to maintain session when you don't put session into dependencies? Is there a way to put session and valid data into same dependencies?

Hello,

Love this repo!

I was trying to modify your example from #22 to check if nested models are round-tripped unnecessarily if contained within the dict I return in my endpoint function. Before I could do that, I had to update the example code in #22 to work with pydantic 2.0 (fastapi==0.100.1, pydantic==2.4.2). I first changed the root_validator to model_validator:

    @model_validator(mode="before")
    @classmethod
    def debug_usage(cls, data: dict):
        print("created pydantic model")

        return data

and when I run the app and hit that endpoint, I see "created pydantic model" once, and do not get "called dict" logged at all.

The dict method is deprecated in favor of model_dump, but if I also override model_dump and model_dump_json:

    def model_dump(self, *args, **kwargs):
        print("called model_dump")
        return super().model_dump(*args, **kwargs)

    def model_dump_json(self, *args, **kwargs):
        print("called model_dump_json")
        return super().model_dump_json(*args, **kwargs)

I don't get any of those "called ..." messages printed. If I use jsonable_encoder on a model in a terminal, I can see it uses model_dump_json, but FastAPI doesn't seem to use any of these!

So my questions are:

1. in recent versions of pydantic and fastapi, what happens if I return an object whose type matches response_model?
2. is this double-encoding problem still a problem? That I only see the object created once makes me think it isn't, but since I can't replicate the full example I'm a but unsure what's going on.

How you guys manage to avoid cirular import errors? I am following this structure but circular import just came out.

How should I start the project, docs says that main.py file will run the app but what will contain that main.py file and where will point?

Example how I did in a recent course
uvicorn app:app --reload --host 0.0.0.0 --port 7070

app.py content:
import fastapi as _fastapi
import fastapi.security as _security
import sqlalchemy.orm as _orm
import schemas as _schemas
import services as _services
from typing import List
from fastapi.middleware.cors import CORSMiddleware

app = _fastapi.FastAPI()

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

@app.post("/api/v1/users")
async def register_user(user: _schemas.UserRequest, db: _orm.Session = _fastapi.Depends(_services.get_db)):
db_user = await _services.get_user_by_email(email=user.email, db=db)
if db_user:
raise _fastapi.HTTPException(status_code=400, detail="Email already exists")
# create user and return token
db_user = await _services.create_user(user=user, db=db)
return await _services.create_token(user=db_user)

@app.post("/api/v1/login")
async def login_user( form_data: _security.OAuth2PasswordRequestForm = _fastapi.Depends(),
db: _orm.Session = _fastapi.Depends(_services.get_db)):
db_user = await _services.login(email=form_data.username, password=form_data.password, db=db)
if not db_user:
raise _fastapi.HTTPException(status_code=401, detail="Wrong login credentials")
return await _services.create_token(db_user)

In this case point to app.py which is llinked to services, so the question is how to structure the main.py in
fastapi-best-practices, where should i point to?

Where this code (@app.post("/api/v1/users") and @app.post("/api/v1/login") ) should be, what namefile should have regarding to fastapi-best-practices?

IMO, there is a need for a starter template to get accustomed to these practice or for better reference. Not compulsory, just my personal opinion.

Filename suggestion

Filenames inside the app/module specific can be example: {module}_service.py.
At some point it would be nice to create a generator for these like in django.
django startapp auth.

Reason

We are already importing them from src.auth import constants as auth_constants it is easier to just change the filename so we don't have to keep track and it doesn't create any conflicts.

PR - #12

Example

Credit: This idea has been inspired by this article and and this comment.

I want to be able to both...

• parse incoming JSON with camelCase naming style, and
• return camelCase JSON in my responses

For this I'm using a custom base model for pretty much all my Pydantic schemas:

from humps import camelize
from pydantic import BaseModel

class BaseSchema(BaseModel):
    class Config:
        # enable sqlalchemy model parsing
        orm_mode = True

        # enable camelCase JSON parsing
        alias_generator = camelize
        allow_population_by_field_name = True

    # enable camelCase json response
    def json(self, *args, **kwargs):
        kwargs.setdefault("by_alias", True)
        return super().json(*args, **kwargs)

(This requires the pyHumps package.)

With this BaseSchema, I can now create pydantic schemas with the intended behavior:

class Device(BaseSchema):
    name: str | None
    serial_number: str

# ingest camelCase
device = Device.parse_raw("""{"name": "Device XYZ", "serialNumber": "XYZ-123-ABC-000"}""")

# return camelCase
print(device.json())
# {"name": "Device XYZ", "serialNumber": "XYZ-123-ABC-000"}

My solution above uses the camelize function from the pyhumps package. Alternatively, you can create the function yourself like this:

def camelize(string: str) -> str:
    string_split = string.split("_")
    return string_split[0] + "".join(word.capitalize() for word in string_split[1:])

Is it possible to update 8 to use pydantic 2+? I'm not pro enough to figure out the migration path for that.

I am planning to use Response Structure like this (https://google.github.io/styleguide/jsoncstyleguide.xml?showone=error#error)

{
  "error": {
    "code": 404,
    "message": "File Not Found",
    "errors": [{
      "domain": "Calendar",
      "reason": "ResourceNotFoundException",
      "message": "File Not Found
    }]
  }
}

Can anyone suggest the best way to handle this?
I handled it for RequestValidationException of Pydantic, but for other Exceptions, I am not sure how to handle it.

For RequestValidationException

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    error_response = {
        "error": {
            "code": status.HTTP_422_UNPROCESSABLE_ENTITY,
            "message": "Request Validation Failed",
            "errors": exc.errors(),
        }
    }
    return JSONResponse(
        content=jsonable_encoder(error_response),
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
    )

First of all - thanks for this beautiful repo!
I noticed that there is no mention of where to store any crud operations in the project structure.
I wonder how you would implement these?
Some implementations I've considered.

1. No separate crud operations - would lead to duplicated code.
2. Following @tiangolo's full stack example structure, there is a separate crud folder. I could create a crud folder for each module.
3. Another implementation I've seen is to declare the crud operations directly in the model's Base class. E.g.:

@classmethod
async def create(cls, **kwargs):
    obj = cls(**kwargs)
    db.add(obj)
    try:
        await db.commit()
    except Exception:
        await db.rollback()
        raise
    return obj

However, this way you don't get any autocompletion.

So far, we're using a separate crud folder for each module, but I would love to hear any other recommendations.

Is it just by Git or you also handle with via API route ?

I recently posted this article about testing FastAPI + Omar + Alembic applications: https://pawamoy.github.io/posts/testing-fastapi-ormar-alembic-apps/

The interesting thing in the post is how each test has access to its own, unique, temporary database :)

That's it, just sharing, feel free to close or comment!

I am following folder structure and following steps to import from other folders like this

from src.dashboard.schemas import Audio

But an error got me, and if i remove the src i got circular import
ModuleNotFoundError: No module named 'src'

Here is my template for using fastapi / docker / poetry. I prefer to use the controller / service / repository pattern. Let me know what you think or if you have any suggestions.

Since you are using sqlalchemy core to make raw sql queries, you might like to use a more maintainable approach to declare and use the tables and columns. ORMs aren't very flexible, but a huge help for type driven development. For example using obj.column is more maintainable than using table.c.column, since the latter is dynamic, and lack type hints.

We can take the middle ground by re-declaring the columns like column = table.c.column, and using the redeclared version everywhere. But that will add a lot of boilerplate code.

One way to avoid boilerplate code is using a factory class like in https://github.com/sayanarijit/sqla-fancy-core which lets us declare the columns almost the same way we declare in the orms, but without the orm magic.

We can also subclass the factory to create custom column type like:

from sqla_fancy_core import TableFactory as _TableFactory

class TableFactory(_TableFactory):
    def col(self, *args, nullable=False, **kwargs):
        kwargs["nullable"] = nullable
        return super().col(*args, **kwargs)

    def foreign_key(
        self,
        name: str,
        ref: str | sa.Column,
        *args,
        onupdate="CASCADE",
        ondelete="CASCADE",
        **kwargs
    ):
        fk = sa.ForeignKey(ref, onupdate=onupdate, ondelete=ondelete)
        return self.col(name, fk, *args, **kwargs)

    def name(self, *args, **kwargs):
        return self.string("name", *args, **kwargs)

    def slug(self, *args, **kwargs):
        return self.text("slug", *args, **kwargs)

I'd be very grateful if you could provide us with the content of database.py because I couldn't understand this part:

from src.database import database, posts, profiles, post_review, products

async def get_posts(
    creator_id: UUID4, *, limit: int = 10, offset: int = 0
) -> list[dict[str, Any]]: 
    select_query = (
        select(
            (
                posts.c.id,
                posts.c.slug,
                posts.c.title,
                func.json_build_object(
                   text("'id', profiles.id"),
                   text("'first_name', profiles.first_name"),
                   text("'last_name', profiles.last_name"),
                   text("'username', profiles.username"),
                ).label("creator"),
            )
        )
        .select_from(posts.join(profiles, posts.c.owner_id == profiles.c.id))
        .where(posts.c.owner_id == creator_id)
        .limit(limit)
        .offset(offset)
        .group_by(
            posts.c.id,
            posts.c.type,
            posts.c.slug,
            posts.c.title,
            profiles.c.id,
            profiles.c.first_name,
            profiles.c.last_name,
            profiles.c.username,
            profiles.c.avatar,
        )
        .order_by(
            desc(coalesce(posts.c.updated_at, posts.c.published_at, posts.c.created_at))
        )
    )
    
    return await database.fetch_all(select_query)

thank you so much

@zhanymkanov thanks for the write up. This article is very useful for me!

In our work, we often deal with a large number of integrations with third-party systems, for this we need to create global clients, what is the most correct way to do this in your opinion?

Issue Title:

Update Links in Readme.md for Smoother Navigation

Issue Content:

In the README.md file, the links at the beginning of the document are currently absolute URLs. When clicked, they reload the page, which can disrupt navigation flow. To improve user experience, I suggest to make the links use anchor links only.

First of all - it's an amazing set of best practices, I also want to share some things that I use:

Project Structure

src could be added to PYTHONPATH to avoid
prefixing every app import with src, IDEs like PyCharm also support that.

Models also could be stored in same package, it's easier to import your models and make sure all modules were executed when you generate your migrations:

src/
  db/
    models/
      __init__.py
      comments.py
      posts.py
      tags.py
    base.py
    dependencies.py

# __init__.py
from . commends import Comment
from . posts import Post
from . tags import Tag

__all__ = ["Comment", "Post", "Tag"]

# Anywhere in the code
from db.models import Post

Continuous Integration

Absolutely use CI in gitlab/github to automate your tests and linters!

Dependency Management

Use poetry instead of requirements.txt, it's awesome!

Custom base model from day 0

This could also be used to enable orm_mode
and set up custom alias_generator if client (for example a JS app) requires it.

Use Starlette's Config object Pydantic BaseSettings!

Pydantic has its own class to manage environment variables:

class AppSettings(BaseSettings):
    class Config:
        env_prefix = "app_"

    domain: str

Thank you for sharing these best practices!

In our projects, we usually define a set of custom exceptions. These are mostly translated into unified error responses.
Eg:

# exceptions.py
class InvalidInputError(Exception):
    error_code = ErrorCode.INVALID_INPUT
    error_message = "Invalid input error"

# Response
400 BadRequest
{
     "error": {
            "error_code": "INVALID_INPUT",
            "error_message": "Missing required field 'abc' ..."
      }
}

I think it would be great if we could have an exception_handlers.py file to handle the mappings from the exceptions to the corresponding error responses.
Eg:

# exceptions.py
class InvalidInputError(Exception):
    error_code = ErrorCode.INVALID_INPUT
    error_message = "Invalid input error"

# exception_handlers.py
def invalid_input_exception_handler(_: Request, exc: InvalidInputError):
    error = ErrorItem(
        error_code=exc.error_code, error_message=exc.error_message
    )
    return JSONResponse(
        status_code=status.HTTP_400_BAD_REQUEST,
        content=jsonable_encoder(ErrorResponse(error=error)),
    )

def register_error_handlers(app: FastAPI) -> None:
    app.add_exception_handler(InvalidInputError, invalid_input_exception_handler)

# main.py
from exception_handlers import register_error_handlers

...
register_error_handlers(app=app)
...

I'm eager to explore the overall arrangement in practical application. Primarily, I'm interested in examining the logical separation of dependencies and services. For instance, while working on the login functionality, I placed JWT within utils and encapsulated user-related operations within a service class, such as get_current_user. However, later on, I realized that this might be categorized as dependencies. I'm familiar with Django but not particularly well-versed in FastAPI's dependencies.

thx

using sys.path.append in every file to access src doesn't seem like such a great idea to me. Maybe I'm doing it wrong i don't know a proper way to do this

Hey, this article is fantastic and inspired a lot to me, appreciate!

for No.19: https://github.com/zhanymkanov/fastapi-best-practices#19-sql-first-pydantic-second, I have some attempts.

Recently I'm working on a small toolkit to handle the nested part.

TLDR: https://github.com/allmonday/pydantic_resolve#demo-2-integrated-with-aiodataloader

We use GraphQL in our project, which is very flexible and allows for easy definition of new fields. When combined with dataloader, it can solve the potential N+1 query problem.

However, as an internal API entry, I feel that GraphQL is too flexible. FastAPI's JSON schema, combined with various client-codegen tools, can reduce a lot of front-end and back-end workload, such as type definitions in front-end.

codegen: https://fastapi.tiangolo.com/advanced/generate-clients/

Therefore, I thought of leveraging the advantages of dataloader and Pydantic:

Pydantic can provide schema, while dataloader can solve the N+1 problem for nested field queries.

        BOOKS_DB = {
            1: [{'name': 'book1'}, {'name': 'book2'}],
            2: [{'name': 'book3'}, {'name': 'book4'}],
        }

        class Book(BaseModel):
            name: str

        class BookLoader(DataLoader):
            async def batch_load_fn(self, keys):
                books = [[Book(**bb) for bb in BOOKS_DB .get(k, [])] for k in keys]
                return books

        book_loader = BookLoader()  

        class Student(BaseModel):
            id: int
            name: str

            books: Tuple[Book, ...] = tuple()
            def resolve_books(self):
                return book_loader.load(self.id)

        # usage
        students = [Student(id=1, name="jack"), Student(id=2, name="mike")]
        results = await resolve(students)

By this way it also simplify the SQL part by moving the nested part into query in loader.

And another bonus is you can reuse the loader anywhere.

Almost all of of here have used some ORM library like Sqlalchemy which is equivalent to how django does its things. I found myself using mongodb as the main database choice for most of my projects. Also in all company projects that Ive been working on since this year, I have not used any sql database. I usually use pymongo and Its always a hassle to configure everything for every new project. What ive done is create a DBConnection class which has all the collections I need as class attributes and possibly some methods does some db operations.

The problem

As you already figured out, Its nothing close to using an ORM, Are there some libraries that support ODM that i can use in fastapi like Sqlalchemy (when dealing with SQL). I mean relationsips, and most important being able to still my favourite mongodb operation which is the aggregate searches.

Really like this work, guys! I want to limit my API to [say] 10 requests/per/minute for each IP. Have you any best practices how to achieve this in FastAPI?