Configuration
First you need to choose orm provider, for this examples we will use SQLAlchemy, and init FastAuthSettings
class.
Let`s init configuration, which we use later.
from fastauth.settings import FastAuthSettings
settings = FastAuthSettings()
There we can share some variables across classes inside library. Also we can extend or override config variables, for example by direct change, or inherit class.
from fastauth.settings import FastAuthSettings
class Settings(FastAuthSettings):
# Override flag to allow inactive users to login
ALLOW_INACTIVE_USERS: bool = False
Models
First of all, we need to create tables inside DB, so let`s implement ORM Models class. FastAuth support sqlalchemy out-the-box, so we just need to inherit ready to use mixins
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from fastauth.contrib.sqlalchemy import BaseUUIDUserModel
class Base(DeclarativeBase):
pass
class User(BaseUUIDUserModel, Base):
pass
Another ID field
You can customize User ID field, you need to inherit BaseUserModel[ID]
class, and set id
field
from fastauth.contrib.sqlalchemy import BaseUserModel
from sqlalchemy.orm import Mapped, mapped_column
class User(BaseUserModel[int]):
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
Now we have User
model. But if you want add support of RBAC, you need to create Role
and Permission
models.
from fastauth.contrib.sqlalchemy import BaseIntRoleModel, BaseIntPermissionModel, BaseRolePermissionRel
class Permission(BaseIntPermissionModel, Base):
pass
class Role(BaseIntRoleModel[Permission], Base):
permisions: Mapped[Permission] = relationship(lazy='selectin')
# Need for sqlalchemy, to indentify many-to-many table
class RolePermissionRel(BaseRolePermissionRel[int, int], Base):
pass
Then we need to upgrade our User
model and add connection between roles and users tables
from fastauth.contrib.sqlalchemy import BaseUUIDUserModel, RBACMixin, BaseUserRoleRel
import uuid
class User(BaseUUIDUserModel,RBACMixin[Role], Base):
roles: Mapped[Role] = relationship(lazy="selectin")
# Need for sqlalchemy, to indentify many-to-many table
class UserRoleRel(BaseUserRoleRel[uuid.UUID, int], Base):
pass
Role and Permission ID field
As for User
model, you can also customize Role
and Permission
id field, by inherit BaseRoleModel
and BasePermissionModel
.
from fastauth.contrib.sqlalchemy import BaseRoleModel, BasePermissionModel
from sqlalchemy.orm import Mapped, mapped_column
import uuid
class Role(BaseRoleModel[uuid.UUID]):
id: Mapped[uuid.UUID] = mapped_column(primary_key=True, autoincrement=True, default=uuid.uuid4)
class Permission(BasePermissionModel[uuid.UUID]):
id: Mapped[uuid.UUID] = mapped_column(primary_key=True, autoincrement=True, default=uuid.uuid4)
For OAuth support, we need create proper model too.
from fastauth.contrib.sqlalchemy import BaseUUIDOAuthAccount
class OAuthAccount(BaseUUIDOAuthAccount):
pass
OAuth ID Field
There is we can customize ID field to, just inherit BaseOAuthModel
class
After, we need to update User
Model, and add proper lazy select for field
from fastauth.contrib.sqlalchemy import BaseUUIDUserModel, RBACMixin, OAuthMixin
import uuid
class User(BaseUUIDUserModel,RBACMixin[Role], OAuthMixin[OAuthAccount], Base):
roles: Mapped[Role] = relationship(lazy="selectin")
oauth_accounts: Mapped[OAuthAccount] = relationship(lazy="joined")
Repositories
To make the library as extensible as possible, we chose the Service-Repository architecture. So we need to implement repository class for every ORM Model.
from fastauth.repositories import IRoleRepository, IUserRepository, IOAuthRepository
from fastauth.contrib.sqlalchemy import (
SQLAlchemyUserRepository,
SQLAlchemyOAuthRepository,
SQLAlchemyRoleRepository,
)
import uuid
class UserRepository(SQLAlchemyUserRepository[User, uuid.UUID]):
model = User
class RoleRepository(SQLAlchemyRoleRepository[Role, int]):
model = Role
class OAuthRepository(SQLAlchemyOAuthRepository[OAuthAccount, uuid.UUID, User]):
model = OAuthAccount
user_model = User
Inside this classes we can override methods to get items from DB. After creation repos, we need to create dependencies, by using FastAPI Depends function, this is very simple.
from fastapi import Depends
from typing import Annotated
async def get_user_repository(session: SessionDep):
return UserRepository(session)
async def get_oauth_repository(session: SessionDep):
return OAuthRepository(session)
async def get_role_repository(session: SessionDep):
return RoleRepository(session)
UserRepoDep = Annotated[IUserRepository, Depends(get_user_repository)]
OAuthRepoDep = Annotated[IOAuthRepository, Depends(get_oauth_repository)]
RoleRepoDep = Annotated[IRoleRepository, Depends(get_role_repository)]
SessionDep
SessionDep
is just annotation for sqlalchemy async session generator
from typing import Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
engine = create_async_engine("<DATABASE_URL>", echo=True)
session_factory = async_sessionmaker(engine, expire_on_commit=False)
async def get_session():
async with session_factory() as session:
try:
yield session
except Exception as e:
await session.rollback()
raise e
finally:
await session.close()
SessionDep = Annotated[AsyncSession, Depends(get_session)]
Token Storage
For user authentication, we need to use tokens. Tokens should be stored somewhere or have a mechanism for verifying authenticity.
For this features we use BaseTokenStorage
class, which handle how and where store tokens. The most simple token storage is jwt, because we do not need to use DB
or Redis to store it physicaly. To work with JWT we need to make dependencies with JWTTokenStorage
class
from fastauth.storage import JWTTokenStorage
def get_auth_storage():
return JWTTokenStorage(settings)
Services
After creating repositories and token storage, we need to implement AuthService class, which handle all business login such as login, token creation, etc.
Inside class we can override some events, such as on_after_register
, on_after_delete
, etc.
from fastauth.services import BaseAuthService
from fastapi import Depends
from typing import Annotated
class AuthService(BaseAuthService):
pass
async def get_auth_service(
user_repo: UserRepoDep,
oauth_repo: OAuthRepoDep,
role_repo: RoleRepoDep,
token_storage: JWTTokenStorage = Depends(get_token_storage),
):
return AuthService(
settings, user_repo, token_storage, oauth_repo=oauth_repo, role_repo=role_repo
)
AuthServiceDep = Annotated[AuthService, Depends(get_auth_service)]
Transport
We need choose throught which transport we get tokens from user in request, it can be Bearer in header or cookie token. To handle this we use BaseTransport
class.
For example we will use CookieTransport
which handle token recieve throught cookies.
from fastauth.transport import CookieTransport
transport = CookieTransport(settings)
FastAuth
Last class which Facade for everything is FastAuth
class. It checks the validity of tokens and whether the user has access to the resource.
from fastauth import FastAuth
security = FastAuth(settings, get_auth_service, transport)
Full SQLAlchemy Example
from fastauth.settings import FastAuthSettings
from pydantic_settings import BaseSettingsModel
class Settings(FastAuthSettings, BaseSettingsModel):
DATABASE_URL: str = "DATABASE_URL"
settings = Settings()
from fastauth.contrib.sqlalchemy import BaseIntRoleModel, BaseIntPermissionModel, BaseRolePermissionRel
class Permission(BaseIntPermissionModel, Base):
pass
class Role(BaseIntRoleModel[Permission], Base):
permisions: Mapped[Permission] = relationship(lazy='selectin')
# Need for sqlalchemy, to indentify many-to-many table
class RolePermissionRel(BaseRolePermissionRel[int, int], Base):
pass
from fastauth.contrib.sqlalchemy import BaseUUIDOAuthAccount
class OAuthAccount(BaseUUIDOAuthAccount):
pass
from fastauth.contrib.sqlalchemy import BaseUUIDUserModel, RBACMixin, OAuthMixin
import uuid
class User(BaseUUIDUserModel,RBACMixin[Role], OAuthMixin[OAuthAccount], Base):
roles: Mapped[Role] = relationship(lazy="selectin")
oauth_accounts: Mapped[OAuthAccount] = relationship(lazy="joined")
from .config import settings
from typing import Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
engine = create_async_engine(settings.DATABASE_URL, echo=True)
session_factory = async_sessionmaker(engine, expire_on_commit=False)
async def get_session():
async with session_factory() as session:
try:
yield session
except Exception as e:
await session.rollback()
raise e
finally:
await session.close()
SessionDep = Annotated[AsyncSession, Depends(get_session)]
from .db import SessionDep
from .models import User, Role, Permission, OAuthAccount
from fastauth.repositories import IRoleRepository, IUserRepository, IOAuthRepository
from fastauth.contrib.sqlalchemy import (
SQLAlchemyUserRepository,
SQLAlchemyOAuthRepository,
SQLAlchemyRoleRepository,
)
import uuid
class UserRepository(SQLAlchemyUserRepository[User, uuid.UUID]):
model = User
class RoleRepository(SQLAlchemyRoleRepository[Role, int]):
model = Role
class OAuthRepository(SQLAlchemyOAuthRepository[OAuthAccount, uuid.UUID, User]):
model = OAuthAccount
user_model = User
from fastapi import Depends
from typing import Annotated
async def get_user_repository(session: SessionDep):
return UserRepository(session)
async def get_oauth_repository(session: SessionDep):
return OAuthRepository(session)
async def get_role_repository(session: SessionDep):
return RoleRepository(session)
UserRepoDep = Annotated[IUserRepository, Depends(get_user_repository)]
OAuthRepoDep = Annotated[IOAuthRepository, Depends(get_oauth_repository)]
RoleRepoDep = Annotated[IRoleRepository, Depends(get_role_repository)]
from .config import settings
from .repositories import UserRepoDep, OAuthRepoDep, RoleRepoDep
from fastauth.storage import JWTTokenStorage
def get_auth_storage():
return JWTTokenStorage(settings)
from fastauth.services import BaseAuthService
from fastapi import Depends
from typing import Annotated
class AuthService(BaseAuthService):
pass
async def get_auth_service(
user_repo: UserRepoDep,
oauth_repo: OAuthRepoDep,
role_repo: RoleRepoDep,
token_storage: JWTTokenStorage = Depends(get_token_storage),
):
return AuthService(
settings, user_repo, token_storage, oauth_repo=oauth_repo, role_repo=role_repo
)
AuthServiceDep = Annotated[AuthService, Depends(get_auth_service)]
from .config import settings
from .services import get_auth_service
from fastauth.transport import CookieTransport
transport = CookieTransport(settings)
security = FastAuth(settings, get_auth_service, transport)