fast-sqlalchemy 0.12.2

Make application with Fastapi and Sqlalchemy with ease

Welcome to fast-sqlalchemy

This project was first made to provide some tools to use Fastapi with SQLAlchemy with ease.

Contents

• Installation
• The database middlewares
  • The DatabaseMiddleware
  • The AutocommitMiddleware
• The event bus
• The yaml config reader
• Pydantic i18n

Installation

Installation using pip:

pip install fast_sqlalchemy

Or with poetry:

poetry add fast_sqlalchemy

The database middlewares

Fast-sqlalchemy provide multiple middlewares to use SQLAlchemy with Fastapi easily

The DatabaseMiddleware

The main middleware is the database middleware which is made to provide a sqlalchemy session accessible throughout your application. We use the ContextVar api of python to have unique session in the context of each request.

To use this middleware you must at first create a Database object where you must pass the url of your database and the engine options of SQLAlchemy:

db = Database(
    URL.create(
        drivername="mariadb+pymysql", username = config["database"]["user"].get(str),
        password=config["database"]["password"].get(str),
        host =config["database"]["host"].get(str), database = config["database"]["name"].get(str)),
    autoflush=False
)

And then register the database middleware:

fastapi = FastAPI()
fastapi.add_middleware(DatabaseMiddleware, db=db)

After that you can have access to the sqlalchemy session of the current request, through the property session of the Database object in the entire application:

db.session.query(User).first()

Note that if you want to have access to a sqlalchemy session outside a request context, you must create a session by using the session contextmanager of the Database object:

with db.session_ctx():
    db.session.query(User).all()

The middleware is actually using this contextmanager for each request.

The AutocommitMiddleware

The auto commit middleware as its name suggest is a middleware which automatically commit at the end of each request. It must be used with the database middleware and must be registered before otherwise it won't work:

fastapi = Fastapi()
fastapi.add_middleware(AutocommitMiddleware, db=db)
fastapi.add_middleware(DatabaseMiddleware, db=db)

The event bus

The event bus provide you a way to emit event in your application and register handlers to handle them. This allows you to create an event-driven architecture for your application.

To use the event bus within your application, you must create at least one event bus and register the event bus middleware to the fastapi middlewares stack

fastapi = FastAPI()
event_bus = LocalEventBus()
fastapi.add_middleware(EventBusMiddleware, buses=[event_bus])

Once the middleware is registered with an event bus you can start creating events and event handlers. Event can be any python object, the most practical way is to create dataclass object:

@dataclass
class EmailChanged:
    email: str

Then you can add an event handler for this event:

@local_event_bus.handler(EmailChanged)
def check_email_uniqueness(e: EmailChanged):
    # some logic
    pass

There are two kinds of handler sync and async handler. Sync handlers are called once the event is emitted, whereas async handlers are called at the end of the current request. To register an async handler is nearly the same as above

@local_event_bus.async_handler(EmailChanged, OtherEvent)
def check_email_uniqueness(e: EmailChanged | OtherEvent):
    # some logic
    pass

Note that a handler can handle multiple types of event

After that you can emit events wherever you want in your Fastapi application:

emit(EmailChanged(email=email))

The database testing class

Fast-sqlalchemy provide a utility class named TestDatabase which can be used to test your Fastapi application with SQLAlchemy with ease. This class allow you to have isolated test by having each test wrapped in a transaction that is rollback at the end of each test, so that each test have a fresh database.

To use it with pytest, you can simply create two fixtures. A primary fixture with a scope of 'session' which will create a connection to the database and create the database if it doesn't exist (A testing database is created with the same name of your application's database prefixed with 'test_'). The testing database is then dropped at the end (You can optionally turn if off).

from my_app import factories

@pytest.fixture(scope="session")
def db_client():
    db_client = TestDatabase(db=app_db, factories_modules=[factories])
    with db_client.start_connection(metadata=metadata):
        yield db_client

Note that this class is compatible with the library factory_boy, you can register as shown in the example above a list of modules which contains your factory classes so that each factory wil be bound to the session provided by the TestDatabase object.

After that you can create a second fixture:

@pytest.fixture()
def sqla_session(db_client):
    with db_client.start_session() as session:
        yield session

This fixture will provide a sqlalchemy session to your tests:

def test_create_user(sqla_session):
    user = UserFactory()
    assert sqla_session.query(User).first().id == user.id

The yaml configuration loader

Fast-sqlalchemy provide a class named Configuration which allow you to have your application's configuration store in yaml files:

ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
config = Configuration(os.path.join(ROOT_DIR, "config"), env_path=os.path.join(ROOT_DIR, ".env"))
config.load_config(config="test")

When you're creating the object you must specify the path of your configuration directory, this directory will contain all of your yaml files. You can also specify a .env file which will be read thanks to the dotenv library. Then you can load these configurations file by calling load_config, you can specify a config name, this config name must match a subdirectory within the configuration directory. This subdirectory should contain yaml files that will be merged with the yaml files present at the root of the configuration directory. This way you can have multiple configurations witch will share the same base configuration. The configuration folder may look like this:

+-- config
|   +-- base.yaml
|   +-- other.yaml
|   +-- test
|    |   +-- base.yaml
|   +-- prod
|    |   +-- base.yaml

Note that you can use your environment variables within your yaml files, these variables will be parsed.

project_name: ${PROJECT_NAME}
secret_key: ${SECRET_KEY}
local: fr_FR

Then you can have access to your configuration within your application like this:

config["general"]["project_name"]

or with the get method witch accept dot-separated notation and a default value as second parameter.

config.get("general.project_name", "default_name")

Note that, if a key is not found in yaml files, as fallback we'll try to find the key in environment or raise a KeyError exception if not present.

Pydantic i18n

This utility class allow you to translate all error messages of pydantic. You can specify a translation for a specific pydantic's error code. for instance:

translations = {
  "fr_FR": {
    "value_error.any_str.max_length": "Ce champs doit faire {0} caractères",
    "value_error.any_str.min_length": "Ce {1} doit faire plus de {0} caractères",
  }
}

You can even organize it with a nested structure:

translations = {
"fr_FR": {
    "value_error": {
      "any_str": {
        "max_length": "Ce champs doit faire {0} caractères",
      }
    } 
  }
}

The error's context can be accessible through the placeholders '{\d}' like the format python's function Then you can use these translations this way:

tr = PydanticI18n(translations, local="fr_FR")

And create a middleware in Fastapi

async def request_validation_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, content=tr.translate(exc.errors()))

Licence

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