SQLAlchemy - My Findings ​

from sqlalchemy.orm import MappedAsDataclass, DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, ForeignKey, func
from datetime import datetime

class Base(MappedAsDataclass, DeclarativeBase):
    pass

class User(Base):
    # Define table name
    __tablename__ = "users"

    # Define primary key. `index=True` is optional when using `primary_key=True`
    # When inheriting from `MappedAsDataclass`, `init=False` is required for primary key to be ignored in `__init__` method
    id: Mapped[int] = mapped_column(primary_key=True, init=False)

    # Define column with length
    name: Mapped[str] = mapped_column(String(30))

    # Auto infer col type from python type
    # `str | None` => `NULL` is allowed
    fullname: Mapped[str | None]

    # `str` => `NOT NULL`
    required_fullname: Mapped[str]

    created_at: Mapped[datetime] = mapped_column(init=False, server_default=func.now())

    addresses: Mapped[list["Address"]] = relationship(init=False, back_populates="user")

class Address(Base):
    __tablename__ = "addresses"

    id: Mapped[int] = mapped_column(primary_key=True, init=False)
    email_address: Mapped[str]
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    user: Mapped["User"] = relationship(init=False, back_populates="addresses")

from sqlalchemy import or_

# Get user by id
user = await db.get(User, 73)

# Get all users (Use this method instead of below three)
statement = select(User)
result = await db.scalars(statement)
users = result.all()

# Get all users (Explicit)
users = db.execute(
    select(User)
).scalars().all()


# Get all users with limit & offset
statement = select(User).offset(skip).limit(limit)
result = await db.scalars(result)
users = result.unique().all()

# Filter user by email
statement = select(User).where(User.email == "john@mail.com")
result = await db.scalars(statement)
user = result.first()
# instead of `.first()` you can also use `.one()` & `.one_or_none()`

# Filter with multiple conditions (AND)
statement = select(User).where(User.email == "john@mail.com").where(User.username == data.username)
result = await db.scalars(statement)
user = result.first()

# Filter with multiple conditions (OR)
statement = select(User).where(or_(User.email == data.email, User.username == data.username))
result = await db.scalars(statement)
user = result.first()

# Order by id
statement = select(User).order_by(User.id.desc())
result = await db.scalars(statement)
users = result.all()

# Get count
statement = select(func.count()).select_from(User)
count = await db.scalar(statement)

statement = select(func.count(User.id))
count = await db.scalar(statement)

# Exist Query
statement = select(exists().where(User.email = "admin@mail.com"))
has_admin_email: bool | None = await db.scalar(statement)

attachments: Mapped[JSONType] = mapped_column(
    JSONB,
    nullable=False,
    default=list,
    server_default=text("'[]'::jsonb"),
)

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)

# Define the function to run after insertion
def after_insert_listener(mapper, connection, target):
    # Your custom logic here
    print("New user added to the DB:", target)

# Attach the event listener to the after_insert event
listens_for(User, 'after_insert', after_insert_listener)

from sqlalchemy.orm import DeclarativeBase, MappedAsDataclass

# Use `MappedAsDataclass` to make models dataclasses and get autocompletion
class Base(MappedAsDataclass, DeclarativeBase):
    pass

class MyModel(BaseModel):
    uploaded_url: HttpUrl

class MyModelDB(Base):
    uploaded_url: Mapped[str]

data = MyModel(uploaded_url="https://example.com")
db.add(MyModelDB(**data.model_dump())) # Error: SQLAlchemy don't accept Url type from pydantic

# Tell pydantic how to serialize the HttpUrl field
class MyModel(BaseModel):
    uploaded_url: Annotated[HttpUrl, PlainSerializer(str)]

db.add(MyModelDB(**data.model_dump())) # Works

from typing import Any
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import text
from sqlalchemy.types import JSON
from sqlalchemy.dialects.postgresql import JSONB

type JSONType = str | int | float | bool | None | dict[str, "JSONType"] | list["JSONType"]

# Use `MappedAsDataclass` to make models dataclasses and get autocompletion
class Base(DeclarativeBase, MappedAsDataclass):
    type_annotation_map = {JSONType: JSON}

class MyModel(Base):
    settings: Mapped[JSONType] = mapped_column(JSONB, default=dict, server_default=text("'{}'::jsonb"))

from datetime import datetime

from sqlalchemy import func
from sqlalchemy.orm import Mapped, mapped_column


class ColIdInt:
    id: Mapped[int] = mapped_column(primary_key=True, init=False)


class ColCreatedAt:
    created_at: Mapped[datetime] = mapped_column(init=False, server_default=func.now())


class ColUpdatedAt:
    updated_at: Mapped[datetime] = mapped_column(
        init=False,
        server_default=func.now(),
        onupdate=func.now(),
        kw_only=True,
    )

# You can use them as mixins
class User(Base, ColIdInt, ColCreatedAt, ColUpdatedAt, ColLastActivityAt):
    __tablename__ = "users"

    name: Mapped[str]