Skip to content


Encode Database is a database that gives you simple asyncio support for a range of databases.

It allows you to make queries using the powerful SQLAlchemy Core expression language and provides support for PostgreSQL, MySQL, and SQLite.

Databases is suitable for integrating against any async Web framework, such as Starlette, Sanic, Responder, Quart, aiohttp, Tornado, or FastAPI.

You could check the documentation for more information.

  • Check that you install requirements

  • Setup the database, and we will use the EncodeDBBackend as a dependency.

import databases
from authx import Authentication, EncodeDBBackend

auth = Authentication(

Here we present a simple PostgreSQL database example on how to use the EncodeDBBackend.

Create/Update/Delete Functions

Here we are creating the functions that will be used to create and update the database,based on the request.

  • Create:
async def create(self, obj: dict) -> UUID:
    uuid = uuid4()
    obj[id] = uuid
    query = self.users.insert()
    await self.database.execute(query=query, values=obj)
    return uuid
  • Update:
 async def update(self, id: UUID, obj: dict) -> bool:
    obj.pop("id", None)
    query = (
        .where( == id)
    res = await self.database.execute(query)
    return bool(res)
  • Delete:
async def delete(self, id: UUID) -> bool:
    query = (
        .where( == id)
    res = await self.database.execute(query)
    return bool(res)

Complex Functions

After you have created the database, you can use the following functions to create, update, and delete users.

  • request Email Confirmation:
async def request_email_confirmation(self, email: str, token_hash: str) -> None:
    query = (
        .values(email=email, token=token_hash)
        .on_conflict_do_update(constraint="email", set_={"token": token_hash})
    await self.database.execute(query)
    return None
  • Confirm Email:
async def confirm_email(self, token_hash: str) -> bool:
    query =
        self.email_confirmations.c.token == token_hash
    email_confirmation = await self.database.fetch_one(query)
    if email_confirmation:
        email = email_confirmation["email"]
        async with self.database.transaction():
            query = (
                .where( == email)
            await self.database.execute(query)
            query = sa.delete(self.email_confirmations).where(
       == email
            await self.database.execute(query)
        return True
        return False

Here we Present different functions to create, update, and delete users, also some complex functions to use it if you would like to extend the functionality.