### PostgreSQL Connection String Example

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/configuration.md

Example connection string for PostgreSQL using the asyncpg driver.

```text
postgresql+asyncpg://user:pass@localhost/db
```

--------------------------------

### MySQL Connection String Example

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/configuration.md

Example connection string for MySQL using the aiomysql driver.

```text
mysql+aiomysql://user:pass@localhost/db
```

--------------------------------

### Get User by Email Example

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/README.md

Demonstrates how to instantiate SQLAlchemyUserDatabase and use the get_by_email method to fetch a user record.

```python
db = SQLAlchemyUserDatabase(session, User)
user = await db.get_by_email("user@example.com")
```

--------------------------------

### Example SQLAlchemyAccessTokenDatabase Configuration

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/configuration.md

Demonstrates setting up an async engine, session factory, and initializing the database with a custom access token model.

```python
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from fastapi_users_db_sqlalchemy.access_token import SQLAlchemyAccessTokenDatabase

# Create async engine
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/dbname")

# Create session factory
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

# Create session
async with AsyncSessionLocal() as session:
    db = SQLAlchemyAccessTokenDatabase(session, AccessToken)
```

--------------------------------

### SQLite Connection String Example

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/configuration.md

Example connection string for SQLite using the aiosqlite driver. Suitable for development and testing.

```text
sqlite+aiosqlite:///./db.db
```

--------------------------------

### Example SQLAlchemyUserDatabase Configuration

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/configuration.md

Demonstrates how to create an async engine, session factory, and then initialize SQLAlchemyUserDatabase with or without OAuth account support.

```python
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from fastapi_users_db_sqlalchemy import SQLAlchemyUserDatabase, SQLAlchemyBaseUserTableUUID

# Create async engine
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/dbname")

# Create session factory
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

# Create session
async with AsyncSessionLocal() as session:
    # Without OAuth
    db = SQLAlchemyUserDatabase(session, User)
    
    # With OAuth
    db = SQLAlchemyUserDatabase(session, User, OAuthAccount)
```

--------------------------------

### FastAPI Integration with SQLAlchemy User Database

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/configuration.md

This example demonstrates a full FastAPI application setup using SQLAlchemyUserDatabase. It includes database engine creation, a dependency for database sessions, and a sample route to retrieve a user.

```python
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from fastapi_users_db_sqlalchemy import SQLAlchemyUserDatabase, SQLAlchemyBaseUserTableUUID

app = FastAPI()

# Database setup
DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

# Dependency
async def get_db_session() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

# Usage in route
@app.get("/users/{user_id}")
async def get_user(
    user_id: str,
    session: AsyncSession = Depends(get_db_session),
):
    db = SQLAlchemyUserDatabase(session, User)
    user = await db.get(user_id)
    return user
```

--------------------------------

### GUID Constructor Example

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/generics.md

Demonstrates how to use the GUID type decorator when defining a SQLAlchemy Column, setting it as the primary key and providing a default UUID value.

```python
from sqlalchemy import Column
from fastapi_users_db_sqlalchemy.generics import GUID
import uuid

id_column = Column(GUID, primary_key=True, default=uuid.uuid4)
```

--------------------------------

### Run Unit Tests with Hatch

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/README.md

Execute all unit tests for the project using the Hatch build tool. Ensure Hatch is installed.

```bash
hatch run test
```

--------------------------------

### Basic User Database Setup with SQLAlchemy

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/REFERENCE.md

Sets up a basic user database using SQLAlchemy ORM with async support. Define your User model inheriting from SQLAlchemyBaseUserTableUUID and create an async engine and sessionmaker. Use SQLAlchemyUserDatabase to interact with the database for user creation.

```python
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from fastapi_users_db_sqlalchemy import (
    SQLAlchemyBaseUserTableUUID,
    SQLAlchemyUserDatabase
)
from sqlalchemy.orm import DeclarativeBase

# Define table
class Base(DeclarativeBase):
    pass

class User(SQLAlchemyBaseUserTableUUID, Base):
    pass

# Create engine and session
engine = create_async_engine("sqlite+aiosqlite:///./test.db")
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

# Use in route
async with AsyncSessionLocal() as session:
    db = SQLAlchemyUserDatabase(session, User)
    user = await db.create({
        "email": "user@example.com",
        "hashed_password": "hashed_password",
    })
```

--------------------------------

### User Database Setup with OAuth Support

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/REFERENCE.md

Integrates OAuth account support into the user database setup. Define an OAuthAccount model inheriting from SQLAlchemyBaseOAuthAccountTableUUID and associate it with the User model. Initialize SQLAlchemyUserDatabase with both User and OAuthAccount models.

```python
from fastapi_users_db_sqlalchemy import SQLAlchemyBaseOAuthAccountTableUUID

class OAuthAccount(SQLAlchemyBaseOAuthAccountTableUUID, Base):
    pass

class User(SQLAlchemyBaseUserTableUUID, Base):
    oauth_accounts: Mapped[list[OAuthAccount]] = relationship(
        "OAuthAccount", lazy="joined"
    )

# Initialize with OAuth
db = SQLAlchemyUserDatabase(session, User, OAuthAccount)
user = await db.add_oauth_account(user, {
    "oauth_name": "google",
    "access_token": "token_xyz",
    "account_id": "google_user_123",
    "account_email": "user@gmail.com",
})
```

--------------------------------

### Usage Example: Defining and Retrieving AccessToken with TIMESTAMPAware

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/generics.md

Shows how to define a table with a TIMESTAMPAware 'created_at' column and how to retrieve records, confirming that the 'created_at' attribute is always timezone-aware.

```python
from datetime import datetime, timezone
from sqlalchemy.orm import Mapped, mapped_column
from fastapi_users_db_sqlalchemy.generics import TIMESTAMPAware, now_utc

# In table definition
class AccessToken(Base):
    __tablename__ = "accesstoken"
    created_at: Mapped[datetime] = mapped_column(
        TIMESTAMPAware(timezone=True),
        nullable=False,
        default=now_utc,
        index=True,
    )

# Retrieving values (all databases)
async with sessionmaker() as session:
    token = await session.get(AccessToken, "token_id")
    
    # token.created_at is always timezone-aware
    print(token.created_at.tzinfo)  # <timezone.utc>
    
    # Safe for comparisons with timezone-aware datetimes
    from datetime import timedelta
    max_age = datetime.now(timezone.utc) - timedelta(hours=24)
    is_valid = token.created_at >= max_age
```

--------------------------------

### GUID Usage in SQLAlchemy Model

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/generics.md

Shows a practical example of integrating the GUID type decorator into a SQLAlchemy model for a user ID. This setup automatically handles database type selection for UUIDs across different SQL dialects.

```python
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
import uuid
from fastapi_users_db_sqlalchemy.generics import GUID

class Base(DeclarativeBase):
    pass

# Simple usage with default
class User(Base):
    __tablename__ = "user"
    id: Mapped[uuid.UUID] = mapped_column(GUID, primary_key=True, default=uuid.uuid4)

# With custom UUID value
user = User(id=uuid.uuid4())

# GUID automatically handles database type selection
# - On PostgreSQL: Native UUID type
# - On MySQL/SQLite: CHAR(36) string storage
```

--------------------------------

### Typical Environment Variables for DATABASE_URL

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/configuration.md

Examples of DATABASE_URL environment variables for different database systems and environments, including PostgreSQL, MySQL, SQLite, and in-memory SQLite.

```bash
# PostgreSQL (Production)
DATABASE_URL=postgresql+asyncpg://user:password@postgres:5432/app_db
```

```bash
# MySQL (Production)
DATABASE_URL=mysql+aiomysql://user:password@mysql:3306/app_db
```

```bash
# SQLite (Development)
DATABASE_URL=sqlite+aiosqlite:///./dev.db
```

```bash
# In-memory SQLite (Testing)
DATABASE_URL=sqlite+aiosqlite:///:memory:
```

--------------------------------

### Access Token Database Setup with SQLAlchemy

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/REFERENCE.md

Sets up a database for storing access tokens using SQLAlchemy. Define an AccessToken model inheriting from SQLAlchemyBaseAccessTokenTableUUID and use SQLAlchemyAccessTokenDatabase to manage token creation.

```python
from fastapi_users_db_sqlalchemy.access_token import (
    SQLAlchemyBaseAccessTokenTableUUID,
    SQLAlchemyAccessTokenDatabase
)

class AccessToken(SQLAlchemyBaseAccessTokenTableUUID, Base):
    pass

async with AsyncSessionLocal() as session:
    db = SQLAlchemyAccessTokenDatabase(session, AccessToken)
    token = await db.create({
        "token": "token_string_here",
        "user_id": user.id,
    })
```

--------------------------------

### FastAPI Users Integration with SQLAlchemy

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/REFERENCE.md

Basic setup for integrating SQLAlchemyUserDatabase with FastAPIUsers. Ensure User, uuid, and auth_backend are defined elsewhere.

```python
from fastapi_users import FastAPIUsers
from fastapi_users.authentication import BearerTransport
from fastapi_users_db_sqlalchemy import SQLAlchemyUserDatabase

transport = BearerTransport(tokenUrl="auth/login")
fastapi_users = FastAPIUsers[User, uuid.UUID](
    user_db,
    [auth_backend],
)
```

--------------------------------

### Define User Model with GUID Type

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/types.md

Example of defining a SQLAlchemy user model using the GUID type for the primary key, with a default UUID generation.

```python
from sqlalchemy.orm import Mapped, mapped_column
from fastapi_users_db_sqlalchemy import SQLAlchemyBaseUserTableUUID
from fastapi_users_db_sqlalchemy.generics import GUID
import uuid

class User(SQLAlchemyBaseUserTableUUID):
    id: Mapped[uuid.UUID] = mapped_column(GUID, primary_key=True, default=uuid.uuid4)
```

--------------------------------

### Example Usage of UUID_ID

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/types.md

Demonstrates how to use the UUID_ID type alias when defining a SQLAlchemyUserDatabase.

```python
from fastapi_users_db_sqlalchemy import UUID_ID, SQLAlchemyUserDatabase

async def example(db: SQLAlchemyUserDatabase[User, UUID_ID]):
    user = await db.get(user_id)
```

--------------------------------

### Catching and Handling SQLAlchemy IntegrityError

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/errors.md

Provides a practical example of how to catch `sqlalchemy.exc.IntegrityError` during user creation and includes the necessary step of rolling back the session to handle the constraint violation gracefully.

```python
from sqlalchemy import exc
from fastapi_users_db_sqlalchemy import SQLAlchemyUserDatabase

db = SQLAlchemyUserDatabase(session, User)

try:
    user = await db.create({
        "email": "user@example.com",
        "hashed_password": "hashed_password",
    })
except exc.IntegrityError:
    print("User with this email already exists")
    # May need to roll back session
    await session.rollback()
```

--------------------------------

### get

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/SQLAlchemyUserDatabase.md

Retrieve a user by their unique ID. Returns the user object if found, otherwise returns None.

```APIDOC
## get

### Description
Retrieve a user by ID.

### Method
GET (Conceptual - SDK method)

### Endpoint
N/A (SDK Method)

### Parameters
#### Path Parameters
- **id** (ID) - Required - User ID to retrieve

### Response
#### Success Response
- **UP** (Optional[UP]) - User instance if found, None otherwise

### Request Example
```python
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi_users_db_sqlalchemy import SQLAlchemyUserDatabase

# Assuming User is your SQLAlchemy model
db = SQLAlchemyUserDatabase(session, User)
user = await db.get(user_id)
if user:
    print(f"Found user: {user.email}")
```
```

--------------------------------

### Define and Use SQLAlchemy Access Token Table

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/access-token-tables.md

Define a custom access token table inheriting from `SQLAlchemyBaseAccessTokenTableUUID` and use `SQLAlchemyAccessTokenDatabase` to manage tokens. This example shows how to create a token for a user and retrieve it with an expiration check.

```python
from sqlalchemy.orm import DeclarativeBase
from fastapi_users_db_sqlalchemy import SQLAlchemyBaseUserTableUUID
from fastapi_users_db_sqlalchemy.access_token import (
    SQLAlchemyBaseAccessTokenTableUUID,
    SQLAlchemyAccessTokenDatabase
)

class Base(DeclarativeBase):
    pass

class AccessToken(SQLAlchemyBaseAccessTokenTableUUID, Base):
    pass

class User(SQLAlchemyBaseUserTableUUID, Base):
    pass

# Create database adapter
db = SQLAlchemyAccessTokenDatabase(session, AccessToken)

# Create a token for a user
token = await db.create({
    "token": "generated_token_string_here",
    "user_id": user.id,
    # created_at is auto-set to now(utc)
})

# Retrieve token with expiration check
from datetime import datetime, timedelta, timezone
max_age = datetime.now(timezone.utc) - timedelta(hours=24)
token = await db.get_by_token("token_string_here", max_age=max_age)
```

--------------------------------

### Custom Type Decorators for SQLAlchemy

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/configuration.md

Shows how to use FastAPI-Users SQLAlchemy type decorators for platform-independent UUID storage (`GUID`) and timezone-aware timestamps (`TIMESTAMPAware`).

```python
from fastapi_users_db_sqlalchemy.generics import GUID, TIMESTAMPAware, now_utc

# GUID type: platform-independent UUID storage
id_column: Mapped[uuid.UUID] = mapped_column(
    GUID,
    primary_key=True,
    default=uuid.uuid4,
)

# TIMESTAMPAware: timezone-aware timestamps on all databases
created_at: Mapped[datetime] = mapped_column(
    TIMESTAMPAware(timezone=True),
    nullable=False,
    default=now_utc,
)
```

--------------------------------

### Type-Safe SQLAlchemy User Database

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/REFERENCE.md

Demonstrates setting up a type-safe SQLAlchemy database instance using Python generics. This ensures that user objects and their IDs are correctly typed throughout the application, enabling methods like 'get' to return properly typed results.

```python
from fastapi_users_db_sqlalchemy import SQLAlchemyUserDatabase, SQLAlchemyBaseUserTableUUID
import uuid

class User(SQLAlchemyBaseUserTableUUID):
    pass

# Type-safe database instance
db: SQLAlchemyUserDatabase[User, uuid.UUID] = SQLAlchemyUserDatabase(session, User)

# Methods are properly typed
user: Optional[User] = await db.get(user_id)
```

--------------------------------

### Usage Example: AccessToken Model

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/types.md

Demonstrates how to use the TIMESTAMPAware type decorator when defining a SQLAlchemy model's column for 'created_at' timestamps. It ensures the 'created_at' field is always stored and retrieved as a UTC-aware datetime.

```python
from datetime import datetime
from sqlalchemy.orm import Mapped, mapped_column
from fastapi_users_db_sqlalchemy.generics import TIMESTAMPAware, now_utc

class AccessToken(SQLAlchemyBaseAccessTokenTableUUID):
    created_at: Mapped[datetime] = mapped_column(
        TIMESTAMPAware(timezone=True),
        nullable=False,
        default=now_utc,
        index=True
    )
```

--------------------------------

### Get User by Email Signature

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/README.md

This is the method signature for retrieving a user by their email address from the SQLAlchemyUserDatabase. It specifies the input parameter and the expected return type.

```python
async def get_by_email(self, email: str) -> Optional[UP]:
```

--------------------------------

### Get Access Token by Token String

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/SQLAlchemyAccessTokenDatabase.md

Retrieves an access token by its token string. Optionally filters by age using a max_age datetime cutoff for expiration checks.

```python
from datetime import datetime, timedelta, timezone
from fastapi_users_db_sqlalchemy.access_token import SQLAlchemyAccessTokenDatabase

db = SQLAlchemyAccessTokenDatabase(session, AccessToken)

# Get token without expiration check
token = await db.get_by_token("token_string_here")

# Get token only if created within last hour
cutoff = datetime.now(timezone.utc) - timedelta(hours=1)
token = await db.get_by_token("token_string_here", max_age=cutoff)

if token:
    print(f"Token is valid, created at: {token.created_at}")
else:
    print("Token not found or expired")
```

--------------------------------

### Pre-built OAuth Account Table with UUID

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/oauth-tables.md

Utilize `SQLAlchemyBaseOAuthAccountTableUUID` for a pre-built OAuth account table model with a UUID primary key and automatic user_id foreign key. This example also shows how to create a `SQLAlchemyUserDatabase` adapter with OAuth support and add an OAuth account to a user.

```python
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from fastapi_users_db_sqlalchemy import (
    SQLAlchemyBaseUserTableUUID,
    SQLAlchemyBaseOAuthAccountTableUUID,
    SQLAlchemyUserDatabase
)

class Base(DeclarativeBase):
    pass

class OAuthAccount(SQLAlchemyBaseOAuthAccountTableUUID, Base):
    pass

class User(SQLAlchemyBaseUserTableUUID, Base):
    oauth_accounts: Mapped[list[OAuthAccount]] = relationship(
        "OAuthAccount", lazy="joined"
    )

# Create database adapter with OAuth support
db = SQLAlchemyUserDatabase(session, User, OAuthAccount)

# Add OAuth account to user
user = await db.add_oauth_account(user, {
    "oauth_name": "google",
    "access_token": "access_token_xyz",
    "expires_at": 1579000751,
    "refresh_token": "refresh_token_xyz",
    "account_id": "google_user_123",
    "account_email": "user@gmail.com",
})
```

--------------------------------

### Token Expiration Validation with max_age

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/access-token-tables.md

Shows how to validate token expiration using the `max_age` parameter. The first example checks if a token is less than 24 hours old, and the second checks for 7-day tokens. Ensure `datetime`, `timedelta`, and `timezone` are imported.

```python
from datetime import datetime, timedelta, timezone

# Check if token is less than 24 hours old
token = await db.get_by_token(
    token_string,
    max_age=datetime.now(timezone.utc) - timedelta(hours=24)
)
```

```python
# For 7-day tokens
token = await db.get_by_token(
    token_string,
    max_age=datetime.now(timezone.utc) - timedelta(days=7)
)
```

--------------------------------

### GUID Type Decorator

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/generics.md

A platform-independent UUID type decorator for SQLAlchemy columns that provides native UUID support across various database backends, falling back to CHAR(36) string storage for MySQL and SQLite.

```APIDOC
## GUID(TypeDecorator)

### Description
Provides native UUID support across all database backends. For PostgreSQL, it uses the native `UUID` type. For MySQL and SQLite, it falls back to `CHAR(36)` string storage. In Python, it always returns `uuid.UUID` objects.

### Constructor
```python
from sqlalchemy import Column
from fastapi_users_db_sqlalchemy.generics import GUID
import uuid

id_column = Column(GUID, primary_key=True, default=uuid.uuid4)
```

### Methods
#### load_dialect_impl(dialect)

Selects the appropriate database-specific type based on the SQLAlchemy dialect.

- **Parameter**: `dialect` (Dialect) - SQLAlchemy dialect instance.
- **Return**: `TypeEngine` - PostgreSQL UUID or CHAR(36) depending on dialect.

#### process_bind_param(value, dialect)

Converts Python UUID to database format for storage, handling dialect-specific requirements.

- **Parameter**: `value` (Optional[uuid.UUID]) - Python UUID object or None.
- **Parameter**: `dialect` (Dialect) - SQLAlchemy dialect instance.
- **Return**: Optional[str] - String representation for storage, or None.

#### process_result_value(value, dialect)

Converts database value back to a Python UUID object.

- **Parameter**: `value` (Optional[str]) - String from database or None.
- **Parameter**: `dialect` (Dialect) - SQLAlchemy dialect instance.
- **Return**: Optional[uuid.UUID] - Python UUID object or None.

### Usage Examples
```python
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
import uuid
from fastapi_users_db_sqlalchemy.generics import GUID

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user"
    id: Mapped[uuid.UUID] = mapped_column(GUID, primary_key=True, default=uuid.uuid4)

user = User(id=uuid.uuid4())
```
```

--------------------------------

### Initialize and Use SQLAlchemyAccessTokenDatabase

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/access-token-tables.md

Demonstrates initializing the database adapter, creating a new token during authentication, validating a token during a request, and revoking a token on logout. Ensure you have a SQLAlchemy session and an AccessToken model defined.

```python
from fastapi_users_db_sqlalchemy.access_token import (
    SQLAlchemyAccessTokenDatabase,
    SQLAlchemyBaseAccessTokenTableUUID
)

# Initialize database adapter
db = SQLAlchemyAccessTokenDatabase(session, AccessToken)

# Create token during authentication
new_token = await db.create({
    "token": "eyJhbGciOiJIUzI1NiIs...",
    "user_id": user.id,
})

# Validate token during request
from datetime import datetime, timedelta, timezone
token = await db.get_by_token(
    request_token,
    max_age=datetime.now(timezone.utc) - timedelta(hours=24)
)

if token:
    user = await user_db.get(token.user_id)
    # Use user for authorization

# Revoke token on logout
await db.delete(token)
```

--------------------------------

### Create Async SQLAlchemy Engines

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/configuration.md

Demonstrates creating asynchronous engines for PostgreSQL, MySQL, and SQLite. For production, PostgreSQL with asyncpg is recommended. Set `echo=True` for SQL logging during development.

```python
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

# PostgreSQL (recommended for production)
engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost:5432/dbname",
    echo=False,  # Set to True for SQL logging
)

# MySQL
engine = create_async_engine(
    "mysql+aiomysql://user:password@localhost:3306/dbname",
)

# SQLite (for development/testing)
engine = create_async_engine(
    "sqlite+aiosqlite:///./test.db",
)
```

--------------------------------

### GUID Type Decorator Implementation

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/generics.md

This code defines the GUID TypeDecorator for SQLAlchemy, which handles native UUID support on PostgreSQL and falls back to CHAR(36) string storage for MySQL and SQLite. It ensures Python UUID objects are consistently loaded and processed.

```python
class GUID(TypeDecorator):
    class UUIDChar(CHAR):
        python_type = UUID4
    
    impl = UUIDChar
    cache_ok = True
    
    def load_dialect_impl(self, dialect):
        if dialect.name == "postgresql":
            return dialect.type_descriptor(UUID())
        else:
            return dialect.type_descriptor(CHAR(36))
    
    def process_bind_param(self, value, dialect):
        if value is None:
            return value
        elif dialect.name == "postgresql":
            return str(value)
        else:
            if not isinstance(value, uuid.UUID):
                return str(uuid.UUID(value))
            else:
                return str(value)
    
    def process_result_value(self, value, dialect):
        if value is None:
            return value
        else:
            if not isinstance(value, uuid.UUID):
                value = uuid.UUID(value)
            return value
```

--------------------------------

### create

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/SQLAlchemyUserDatabase.md

Create a new user in the database with the provided attributes. Returns the newly created user instance.

```APIDOC
## create

### Description
Create a new user in the database.

### Method
POST (Conceptual - SDK method)

### Endpoint
N/A (SDK Method)

### Parameters
#### Path Parameters
- **create_dict** (dict[str, Any]) - Required - Dictionary of user attributes. Must include required fields like `email` and `hashed_password`

### Response
#### Success Response
- **UP** (UP) - The created user instance with generated ID

### Throws
- **sqlalchemy.exc.IntegrityError** - if email already exists (violates unique constraint)

### Request Example
```python
db = SQLAlchemyUserDatabase(session, User)
new_user = await db.create({
    "email": "newuser@example.com",
    "hashed_password": "hashed_pwd_here",
    "is_active": True,
    "is_superuser": False,
    "is_verified": False,
})
print(f"Created user with ID: {new_user.id}")
```
```

--------------------------------

### Create Basic User with SQLAlchemy

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/README.md

Use SQLAlchemyUserDatabase to create a new user with email and hashed password. Ensure the session and User model are correctly configured.

```python
db = SQLAlchemyUserDatabase(session, User)
user = await db.create({
    "email": "user@example.com",
    "hashed_password": "hashed_pwd",
})
```

--------------------------------

### Complete User Creation with OAuth

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/REFERENCE.md

Demonstrates how to create a user and add an OAuth account using SQLAlchemyUserDatabase. This is useful for integrating third-party authentication providers.

```python
from fastapi_users_db_sqlalchemy import SQLAlchemyUserDatabase

db = SQLAlchemyUserDatabase(session, User, OAuthAccount)

# Create user
user = await db.create({
    "email": "user@example.com",
    "hashed_password": "hashed_password",
})

# Add OAuth account
user = await db.add_oauth_account(user, {
    "oauth_name": "google",
    "access_token": "google_token",
    "account_id": "google_user_123",
    "account_email": "user@gmail.com",
})

# Update OAuth token on refresh
user = await db.update_oauth_account(user, user.oauth_accounts[0], {
    "access_token": "new_google_token",
    "expires_at": new_expiry,
})
```

--------------------------------

### Configure Async Session Factory

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/configuration.md

Sets up an asynchronous session factory with recommended settings, including `expire_on_commit=False` to ensure loaded objects remain valid after commits.

```python
# Create session factory with recommended settings
AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,  # Important: keeps loaded objects valid after commit
    nullable=True,
)
```

--------------------------------

### Implement GUID Type Decorator for SQLAlchemy

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/types.md

SQLAlchemy type decorator for platform-independent UUID handling. Falls back to CHAR(36) for non-PostgreSQL databases.

```python
class GUID(TypeDecorator):
    """Platform-independent GUID type."""
    
    class UUIDChar(CHAR):
        python_type = UUID4
    
    impl = UUIDChar
    cache_ok = True
    
    def load_dialect_impl(self, dialect):
        if dialect.name == "postgresql":
            return dialect.type_descriptor(UUID())
        else:
            return dialect.type_descriptor(CHAR(36))
    
    def process_bind_param(self, value, dialect):
        # Converts Python UUID to database format
    
    def process_result_value(self, value, dialect):
        # Converts database format back to Python UUID
```

--------------------------------

### create

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/SQLAlchemyAccessTokenDatabase.md

Creates a new access token in the database.

```APIDOC
## create(create_dict: dict[str, Any]) -> AP

### Description
Create a new access token in the database.

### Parameters
#### Path Parameters
- **create_dict** (dict[str, Any]) - Required - Dictionary containing token data. Must include `token` (str) and `user_id` (ID). `created_at` is auto-set if not provided

### Response
#### Success Response (200)
- **AP** (AP) - The created access token instance

### Throws
- **sqlalchemy.exc.IntegrityError** - if token already exists (token is primary key)

### Request Example
```python
from fastapi_users_db_sqlalchemy.access_token import SQLAlchemyAccessTokenDatabase
import uuid

db = SQLAlchemyAccessTokenDatabase(session, AccessToken)

token = await db.create({
    "token": "token_string_here",
    "user_id": uuid.UUID("12345678-1234-5678-1234-567812345678")
})
print(f"Token created with ID: {token.token}")
```
```

--------------------------------

### SQLAlchemyUserDatabase Constructor

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/SQLAlchemyUserDatabase.md

Instantiate SQLAlchemyUserDatabase with an async session, user table model, and an optional OAuth account table model.

```python
SQLAlchemyUserDatabase(
    session: AsyncSession,
    user_table: type[UP],
    oauth_account_table: Optional[type[SQLAlchemyBaseOAuthAccountTable]] = None,
)
```

--------------------------------

### SQLAlchemyAccessTokenDatabase Constructor

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/configuration.md

Initializes the SQLAlchemyAccessTokenDatabase with an async session and the access token table model.

```python
SQLAlchemyAccessTokenDatabase(
    session: AsyncSession,
    access_token_table: type[AP],
)
```

--------------------------------

### Integrate OAuth with SQLAlchemyUserDatabase

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/oauth-tables.md

Demonstrates initializing `SQLAlchemyUserDatabase` with OAuth support and common operations like retrieving a user by OAuth credentials and updating OAuth credentials.

```python
from fastapi_users_db_sqlalchemy import SQLAlchemyUserDatabase

# Initialize with OAuth support
db = SQLAlchemyUserDatabase(session, User, OAuthAccount)

# Retrieve user by OAuth credentials
user = await db.get_by_oauth_account("google", "google_user_123")

# Update OAuth credentials when they refresh
user = await db.update_oauth_account(user, oauth_account, {
    "access_token": "new_token",
    "expires_at": new_expiry_timestamp,
})
```

--------------------------------

### SQLAlchemyAccessTokenDatabase Constructor

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/SQLAlchemyAccessTokenDatabase.md

Initializes the SQLAlchemyAccessTokenDatabase with a session and the access token table model.

```APIDOC
## SQLAlchemyAccessTokenDatabase(session: AsyncSession, access_token_table: type[AP])

### Description
Initializes the database adapter for storing and managing access tokens using SQLAlchemy.

### Parameters
#### Path Parameters
- **session** (AsyncSession) - Required - SQLAlchemy async session instance for database operations
- **access_token_table** (type[AP]) - Required - SQLAlchemy access token model class inheriting from `SQLAlchemyBaseAccessTokenTable[ID]`
```

--------------------------------

### SQLAlchemyUserDatabase Constructor

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/SQLAlchemyUserDatabase.md

Initializes the SQLAlchemyUserDatabase with a SQLAlchemy async session, user model class, and an optional OAuth account model class.

```APIDOC
## SQLAlchemyUserDatabase Constructor

### Description
Initializes the SQLAlchemyUserDatabase with a SQLAlchemy async session, user model class, and an optional OAuth account model class.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Method
Constructor

### Endpoint
N/A

### Parameters
- **session** (`AsyncSession`) - Required - SQLAlchemy async session instance, used for all database operations
- **user_table** (`type[UP]`) - Required - SQLAlchemy user model class inheriting from `SQLAlchemyBaseUserTable[ID]`
- **oauth_account_table** (`Optional[type[SQLAlchemyBaseOAuthAccountTable]]`) - Optional - Optional SQLAlchemy OAuth account model class for OAuth operations
```

--------------------------------

### now_utc() Utility Function

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/generics.md

Helper function for getting current UTC time. This function is intended to be used as a default value for SQLAlchemy timestamp columns, ensuring that timestamps are consistently recorded in UTC.

```APIDOC
## now_utc()

### Description
Helper function for getting current UTC time. Provides a clean default value function for timestamp columns, used as a SQLAlchemy column default.

### Function Signature
`now_utc() -> datetime`

### Import Path
`fastapi_users_db_sqlalchemy.generics`

### Return Value
`datetime` — Current date and time in UTC timezone

### Usage Example
```python
from datetime import datetime
from sqlalchemy.orm import Mapped, mapped_column
from fastapi_users_db_sqlalchemy.generics import TIMESTAMPAware, now_utc

class AccessToken(Base):
    __tablename__ = "accesstoken"
    
    # Auto-set to current UTC time on creation
    created_at: Mapped[datetime] = mapped_column(
        TIMESTAMPAware(timezone=True),
        nullable=False,
        default=now_utc,  # Pass the function itself, not the result
    )

# Creating a record
token = AccessToken(token="xyz")
# token.created_at is automatically set to current UTC time
```

### Important Note on Function vs. Result
When used as a column default, pass the function object (`default=now_utc`), not the result of calling it (`default=now_utc()`). SQLAlchemy calls the function for each new row.
```

--------------------------------

### Typical Authentication Flow with SQLAlchemyAccessTokenDatabase

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/SQLAlchemyAccessTokenDatabase.md

Demonstrates the common lifecycle of an access token: creation during login, validation during request processing, and deletion during logout. Ensure `SQLAlchemyAccessTokenDatabase` is initialized with a session and token model.

```python
from fastapi_users_db_sqlalchemy.access_token import SQLAlchemyAccessTokenDatabase
from datetime import datetime, timedelta, timezone

db = SQLAlchemyAccessTokenDatabase(session, AccessToken)

# During login: create a token
new_token = await db.create({
    "token": generate_token(),
    "user_id": user.id,
    # created_at is auto-set to now(utc)
})

# During request validation: check token validity
max_age = datetime.now(timezone.utc) - timedelta(hours=24)
token = await db.get_by_token(request_token, max_age=max_age)

if token:
    # Token is valid and fresh
    user = await user_db.get(token.user_id)
else:
    # Token is invalid or expired
    raise AuthenticationError()

# During logout: delete the token
await db.delete(token)
```

--------------------------------

### Format Code with Hatch

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/README.md

Apply code formatting using isort and black by running this command with Hatch. This ensures consistent code style.

```bash
hatch run lint
```

--------------------------------

### Token-Based Authentication Flow

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/REFERENCE.md

Illustrates the process of creating, validating, and deleting authentication tokens. Use this for managing user sessions and ensuring token validity.

```python
from datetime import datetime, timedelta, timezone

# Create token
token = await token_db.create({
    "token": generate_token(),
    "user_id": user.id,
})

# Validate token (with 24-hour expiration)
max_age = datetime.now(timezone.utc) - timedelta(hours=24)
token = await token_db.get_by_token(request_token, max_age=max_age)

if token:
    user = await user_db.get(token.user_id)
    # User is authenticated
else:
    # Token invalid or expired

# Logout
await token_db.delete(token)
```

--------------------------------

### Create New Access Token

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/SQLAlchemyAccessTokenDatabase.md

Creates a new access token in the database. Requires 'token' and 'user_id' in the create_dict. 'created_at' is auto-set if not provided. Throws IntegrityError if the token already exists.

```python
from fastapi_users_db_sqlalchemy.access_token import SQLAlchemyAccessTokenDatabase
import uuid

db = SQLAlchemyAccessTokenDatabase(session, AccessToken)

token = await db.create({
    "token": "token_string_here",
    "user_id": uuid.UUID("12345678-1234-5678-1234-567812345678"),
})
print(f"Token created with ID: {token.token}")
```

--------------------------------

### Import Core SQLAlchemy Database Classes

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/README.md

Imports the main user and access token database classes, along with base table definitions and utility types from the fastapi-users-db-sqlalchemy library.

```python
from fastapi_users_db_sqlalchemy import (
    SQLAlchemyUserDatabase,
    SQLAlchemyBaseUserTable,
    SQLAlchemyBaseUserTableUUID,
    SQLAlchemyBaseOAuthAccountTable,
    SQLAlchemyBaseOAuthAccountTableUUID,
    UUID_ID,
)

from fastapi_users_db_sqlalchemy.access_token import (
    SQLAlchemyAccessTokenDatabase,
    SQLAlchemyBaseAccessTokenTable,
    SQLAlchemyBaseAccessTokenTableUUID,
)

from fastapi_users_db_sqlalchemy.generics import (
    GUID,
    TIMESTAMPAware,
    now_utc,
)
```

--------------------------------

### Configure SQLAlchemy Engine with Environment Variable

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/configuration.md

This snippet shows how to create an asynchronous SQLAlchemy engine using a database URL read from the environment. It provides a default for development environments.

```python
import os
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

# Read from environment
DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "sqlite+aiosqlite:///./test.db"  # Development default
)

engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
```

--------------------------------

### Initializing SQLAlchemyUserDatabase with OAuth Table

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/errors.md

Provide the `oauth_account_table` parameter when initializing `SQLAlchemyUserDatabase` to enable OAuth operations. This prevents `NotImplementedError`.

```python
from fastapi_users_db_sqlalchemy import SQLAlchemyUserDatabase

# Correct: OAuth table provided
db = SQLAlchemyUserDatabase(session, User, OAuthAccount)
user = await db.add_oauth_account(user, oauth_data)  # Works

# Wrong: OAuth table not provided
db = SQLAlchemyUserDatabase(session, User)
user = await db.add_oauth_account(user, oauth_data)  # Raises NotImplementedError
```

--------------------------------

### SQLAlchemyAccessTokenDatabase Methods

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/INDEX.md

Reference for methods available on the SQLAlchemyAccessTokenDatabase adapter for managing access tokens.

```APIDOC
## SQLAlchemyAccessTokenDatabase

### get_by_token()

Retrieves an access token by its token value, optionally considering a maximum age.

- **Parameters**
  - `token`: The token string to look up.
  - `max_age`: The maximum age in seconds for the token to be considered valid.
- **Return**
  - `Optional[Token]`: The token object if found and valid, otherwise None.

### create()

Creates a new access token.

- **Parameters**
  - `create_dict`: A dictionary containing access token creation data.
- **Return**
  - `Token`: The newly created token object.
- **Throws**
  - `IntegrityError`: If a unique constraint is violated.

### update()

Updates an existing access token.

- **Parameters**
  - `token`: The token object to update.
  - `update_dict`: A dictionary containing the fields to update.
- **Return**
  - `Token`: The updated token object.

### delete()

Deletes an access token.

- **Parameters**
  - `token`: The token object to delete.
- **Return**
  - `None`
```

--------------------------------

### SQLAlchemy Connection String Format

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/configuration.md

Use this format to specify your database connection string. Ensure the dialect and driver match your chosen backend.

```text
dialect+driver://username:password@host:port/database
```

--------------------------------

### Constructor Usage for TIMESTAMPAware

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/generics.md

Demonstrates how to use the TIMESTAMPAware type in a SQLAlchemy mapped column definition. It ensures timezone support is enabled and sets a default value using now_utc.

```python
from datetime import datetime
from sqlalchemy.orm import Mapped, mapped_column
from fastapi_users_db_sqlalchemy.generics import TIMESTAMPAware, now_utc

created_at: Mapped[datetime] = mapped_column(
    TIMESTAMPAware(timezone=True),
    nullable=False,
    default=now_utc,
)
```

--------------------------------

### SQLAlchemyUserDatabase Methods

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/INDEX.md

Reference for methods available on the SQLAlchemyUserDatabase adapter for managing user data.

```APIDOC
## SQLAlchemyUserDatabase

### get()

Retrieves a user by their ID.

- **Parameters**
  - `id`: The ID of the user to retrieve.
- **Return**
  - `Optional[User]`: The user object if found, otherwise None.

### get_by_email()

Retrieves a user by their email address.

- **Parameters**
  - `email`: The email address of the user to retrieve.
- **Return**
  - `Optional[User]`: The user object if found, otherwise None.

### get_by_oauth_account()

Retrieves a user associated with a specific OAuth account.

- **Parameters**
  - `oauth`: The OAuth provider name.
  - `account_id`: The account ID from the OAuth provider.
- **Return**
  - `Optional[User]`: The user object if found, otherwise None.
- **Throws**
  - `NotImplementedError`: If the method is not implemented.

### create()

Creates a new user.

- **Parameters**
  - `create_dict`: A dictionary containing user creation data.
- **Return**
  - `User`: The newly created user object.
- **Throws**
  - `IntegrityError`: If a unique constraint is violated (e.g., duplicate email).

### update()

Updates an existing user.

- **Parameters**
  - `user`: The user object to update.
  - `update_dict`: A dictionary containing the fields to update.
- **Return**
  - `User`: The updated user object.

### delete()

Deletes a user.

- **Parameters**
  - `user`: The user object to delete.
- **Return**
  - `None`

### add_oauth_account()

Adds an OAuth account to an existing user.

- **Parameters**
  - `user`: The user object to associate the OAuth account with.
  - `create_dict`: A dictionary containing OAuth account creation data.
- **Return**
  - `User`: The updated user object.
- **Throws**
  - `NotImplementedError`: If the method is not implemented.

### update_oauth_account()

Updates an existing OAuth account for a user.

- **Parameters**
  - `user`: The user object.
  - `oauth`: The OAuth provider name.
  - `update_dict`: A dictionary containing the fields to update.
- **Return**
  - `User`: The updated user object.
- **Throws**
  - `NotImplementedError`: If the method is not implemented.
```

--------------------------------

### Create Access Token with SQLAlchemy

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/README.md

Utilize SQLAlchemyAccessTokenDatabase to store JWT or other tokens. Requires the session, AccessToken model, and the user ID to associate the token with.

```python
db = SQLAlchemyAccessTokenDatabase(session, AccessToken)
token = await db.create({
    "token": "token_string",
    "user_id": user.id,
})
```

--------------------------------

### Integration with User Database

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/oauth-tables.md

Demonstrates how to integrate OAuth account tables with SQLAlchemyUserDatabase for authentication and management.

```APIDOC
## Integration with User Database

OAuth account tables are used with `SQLAlchemyUserDatabase` to enable OAuth authentication:

```python
from fastapi_users_db_sqlalchemy import SQLAlchemyUserDatabase

# Initialize with OAuth support
db = SQLAlchemyUserDatabase(session, User, OAuthAccount)

# Retrieve user by OAuth credentials
user = await db.get_by_oauth_account("google", "google_user_123")

# Update OAuth credentials when they refresh
user = await db.update_oauth_account(user, oauth_account, {
    "access_token": "new_token",
    "expires_at": new_expiry_timestamp,
})
```

### Cascade Delete Behavior

When using `SQLAlchemyBaseOAuthAccountTableUUID`, the foreign key constraint includes `ondelete="cascade"`. This means:
- Deleting a user automatically deletes all associated OAuth accounts
- No orphaned OAuth records remain in the database
```

--------------------------------

### Create New User with SQLAlchemy

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/SQLAlchemyUserDatabase.md

Creates a new user in the database with the provided attributes. Raises an IntegrityError if the email already exists.

```python
db = SQLAlchemyUserDatabase(session, User)
new_user = await db.create({
    "email": "newuser@example.com",
    "hashed_password": "hashed_pwd_here",
    "is_active": True,
    "is_superuser": False,
    "is_verified": False,
})
print(f"Created user with ID: {new_user.id}")
```

--------------------------------

### Add OAuth Account to User

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/SQLAlchemyUserDatabase.md

Associates an OAuth account with a user instance. Requires the user object and a dictionary containing OAuth account details. Throws NotImplementedError if oauth_account_table was not provided.

```python
async def add_oauth_account(self, user: UP, create_dict: dict[str, Any]) -> UP:
    pass
```

```python
db = SQLAlchemyUserDatabase(session, User, OAuthAccount)
user = await db.get(user_id)
user = await db.add_oauth_account(user, {
    "oauth_name": "google",
    "access_token": "google_access_token_xyz",
    "expires_at": 1579000751,
    "refresh_token": "google_refresh_token_xyz",
    "account_id": "google_user_123",
    "account_email": "user@gmail.com",
})
```

--------------------------------

### Add OAuth Account with SQLAlchemy

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/README.md

Integrate OAuth accounts by initializing SQLAlchemyUserDatabase with the OAuthAccount table. This allows linking external provider credentials to a user.

```python
db = SQLAlchemyUserDatabase(session, User, OAuthAccount)
user = await db.add_oauth_account(user, {
    "oauth_name": "google",
    "access_token": "token_xyz",
    "account_id": "google_user_123",
    "account_email": "user@gmail.com",
})
```

--------------------------------

### Custom User Fields with SQLAlchemy

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/REFERENCE.md

Shows how to extend the default user model with custom fields like first name, last name, and phone number. This is essential for tailoring the user model to specific application needs.

```python
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import String
from fastapi_users_db_sqlalchemy import SQLAlchemyBaseUserTableUUID

class User(SQLAlchemyBaseUserTableUUID, Base):
    # Standard fields inherited from base
    
    # Custom fields
    first_name: Mapped[str] = mapped_column(String(255), nullable=True)
    last_name: Mapped[str] = mapped_column(String(255), nullable=True)
    phone: Mapped[str] = mapped_column(String(20), nullable=True)
```

--------------------------------

### Async Session Dependency for FastAPI

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/README.md

Provides an asynchronous database session for use in FastAPI route handlers. Ensure AsyncSessionLocal is properly configured.

```python
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

async def get_session() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

# In route handlers
async def get_user(session: AsyncSession = Depends(get_session)):
    db = SQLAlchemyUserDatabase(session, User)
    # Use db for user operations
```

--------------------------------

### Retrieve User by Email with SQLAlchemy

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/SQLAlchemyUserDatabase.md

Fetches a user from the database using their email address, performing a case-insensitive search. Returns the user object if found, otherwise None.

```python
db = SQLAlchemyUserDatabase(session, User)
user = await db.get_by_email("user@example.com")
# Also works with different case:
user = await db.get_by_email("USER@EXAMPLE.COM")
```

--------------------------------

### FastAPI Users SQLAlchemy DB Import Paths

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/INDEX.md

Import common classes for user and access token databases, as well as type decorators and utilities. Ensure you have the necessary classes imported for your specific use case.

```python
# Main classes
from fastapi_users_db_sqlalchemy import (
    SQLAlchemyUserDatabase,
    SQLAlchemyBaseUserTable,
    SQLAlchemyBaseUserTableUUID,
    SQLAlchemyBaseOAuthAccountTable,
    SQLAlchemyBaseOAuthAccountTableUUID,
    UUID_ID,
)

# Access token classes
from fastapi_users_db_sqlalchemy.access_token import (
    SQLAlchemyAccessTokenDatabase,
    SQLAlchemyBaseAccessTokenTable,
    SQLAlchemyBaseAccessTokenTableUUID,
)

# Type decorators and utilities
from fastapi_users_db_sqlalchemy.generics import (
    GUID,
    TIMESTAMPAware,
    now_utc,
)
```

--------------------------------

### User Table Schema

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/REFERENCE.md

Defines the schema for the 'user' table, including fields for ID, email, password, and user status flags. An index is created on the lowercased email for efficient lookups.

```sql
CREATE TABLE "user" (
    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
    email VARCHAR(320) NOT NULL UNIQUE,
    hashed_password VARCHAR(1024) NOT NULL,
    is_active BOOLEAN NOT NULL DEFAULT TRUE,
    is_superuser BOOLEAN NOT NULL DEFAULT FALSE,
    is_verified BOOLEAN NOT NULL DEFAULT FALSE
);

CREATE INDEX ix_user_email ON "user"(LOWER(email));
```

--------------------------------

### Retrieve User by ID with SQLAlchemy

Source: https://github.com/fastapi-users/fastapi-users-db-sqlalchemy/blob/main/_autodocs/api-reference/SQLAlchemyUserDatabase.md

Fetches a user from the database using their unique ID. Returns the user object if found, otherwise None.

```python
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi_users_db_sqlalchemy import SQLAlchemyUserDatabase

# Assuming User is your SQLAlchemy model
db = SQLAlchemyUserDatabase(session, User)
user = await db.get(user_id)
if user:
    print(f"Found user: {user.email}")
```