fastapi-injectable 0.1.0

Allow you to painlessly use dependency injection mechanism (`Depends`) of FastAPI outside the FastAPI routes

FastAPI Injectable

A lightweight package that lets you use FastAPI's dependency injection anywhere - not just in route handlers. Perfect for CLI tools, background tasks, and more.

Overview

fastapi-injectable is a lightweight package that enables seamless use of FastAPI's dependency injection system outside of route handlers. It solves a common pain point where developers need to reuse FastAPI dependencies in non-HTTP contexts like CLI tools, background tasks, or scheduled jobs.

Quick Start

Key features:

• Flexible Injection APIs: Choose between decorator or function-based approaches

  from fastapi_injectable.decorator import injectable
  from fastapi_injectable.util import get_injected_obj
  
  # 1. Decorator approach
  @injectable
  def process_data(db: Annotated[Database, Depends(get_db)]):
      return db.query()
  result1 = process_data()
  
  # 2. Function-wrapper approach
  def process_data(db: Annotated[Database, Depends(get_db)]):
      return db.query()
  
  injectable_process_data = injectable(process_data)
  result2 = injectable_process_data()
  
  # 3. Use the utility
  def process_data(db: Annotated[Database, Depends(get_db)]):
      return db.query()
  
  result3 = get_injected_obj(process_data)
  
  # They are all the same!
  assert result1 == result2 == result3
  

• Support for Both Sync and Async: Works seamlessly with both synchronous and asynchronous code

  from fastapi_injectable.decorator import injectable
  
  def get_service():
      return Service()
  
  @injectable
  async def async_task(service: Annotated[Service, Depends(get_service)]):
      await service.process()
  

• Controlled Resource Management: Explicit cleanup of dependencies through utility functions

  from fastapi_injectable.decorator import injectable
  from fastapi_injectable.util import cleanup_all_exit_stacks, cleanup_exit_stack_of_func
  
  # Define a dependency with cleanup
  def get_db() -> Generator[Database, None, None]:
      db = Database()
      yield db
      db.cleanup()  # Called when cleanup functions are invoked
  
  # Use the dependency
  @injectable
  def process_data(db: Annotated[Database, Depends(get_db)]):
      return db.query()
  
  # Cleanup options
  await cleanup_exit_stack_of_func(process_data)  # Cleanup specific function's resources
  await cleanup_all_exit_stacks()  # Cleanup all resources
  

• Dependency Caching: Optional caching of resolved dependencies for better performance

  from fastapi_injectable.decorator import injectable
  
  def get_mayor() -> Mayor:
      return Mayor()
  
  def get_capital(mayor: Annotated[Mayor, Depends(get_mayor)]) -> Capital:
      return Capital(mayor)
  
  @injectable
  def get_country(capital: Annotated[Capital, Depends(get_capital)]) -> Country:
      return Country(capital)
  
  # With caching (default), all instances share the same dependencies
  country_1 = get_country()
  country_2 = get_country()
  country_3 = get_country()
  assert country_1.capital is country_2.capital is country_3.capital
  assert country_1.capital.mayor is country_2.capital.mayor is country_3.capital.mayor
  
  # Without caching, new instances are created each time
  @injectable(use_cache=False)
  def get_country(capital: Annotated[Capital, Depends(get_capital)]) -> Country:
      return Country(capital)
  
  country_1 = get_country()
  country_2 = get_country()
  country_3 = get_country()
  assert country_1.capital is not country_2.capital is not country_3.capital
  assert country_1.capital.mayor is not country_2.capital.mayor is not country_3.capital.mayor
  

• Graceful Shutdown: Built-in utilities for proper cleanup during application shutdown

  from fastapi_injectable import setup_graceful_shutdown
  
  setup_graceful_shutdown()  # Handles SIGTERM and SIGINT
  

This package is particularly useful for:

• Background task workers
• CLI applications
• Scheduled jobs
• Test fixtures
• Any non-HTTP context where you want to leverage FastAPI's dependency injection

Table of Content

• FastAPI Injectable
  • Overview
  • Quick Start
  • Requirements
  • Installation
  • Real-world Examples
    • 1. Using Depends in in-house background worker
  • Contributing
  • License
  • Issues
  • Credits
  • Bonus

Requirements

• Python 3.10 or higher
• FastAPI 0.112.4 or higher

Installation

You can install fastapi-injectable via pip from PyPI:

$ pip install fastapi-injectable

Real-world Examples

1. Using Depends in in-house background worker

Here's a practical example of using fastapi-injectable in a background worker that processes messages.

You can find the complete example with more details in the examples/worker directory.

This example demonstrates several key patterns for using dependency injection in background workers:

1. Fresh Dependencies per Message:

   • Each message gets a fresh set of dependencies through _init_as_consumer()
   • This ensures clean state for each message, similar to how FastAPI handles HTTP requests

2. Proper Resource Management:

   • Dependencies with cleanup needs (like database connections) are properly handled
   • Cleanup code in generators runs when cleanup_exit_stack_of_func() is called
   • Cache is cleared between messages to prevent memory leaks

3. Graceful Shutdown:

   • setup_graceful_shutdown() ensures resources are cleaned up on program termination
   • Handles both SIGTERM and SIGINT signals

This pattern is particularly useful for:

• Message queue consumers
• Batch processing jobs
• Long-running background tasks
• Any scenario where you need FastAPI-style dependency injection in a worker process

Contributing

Contributions are very welcome. To learn more, see the Contributor Guide.

License

Distributed under the terms of the MIT license, fastapi-injectable is free and open source software.

Issues

If you encounter any problems, please file an issue along with a detailed description.

Credits

1. This project was generated from @cjolowicz's Hypermodern Python Cookiecutter template.
2. Thanks to @barapa's initiation, his work inspires me to create this project.

Bonus

My blog posts about the prototype of this project:

1. Easily Reusing Depends Outside FastAPI Routes
2. 在 FastAPI Routes 以外無痛複用 Depends 的方法