95 lines
3.1 KiB
Python
95 lines
3.1 KiB
Python
"""Minimal Postgres example using sqlmodel-pg-kit.
|
|
|
|
Prerequisites:
|
|
1. Start the dockerized Postgres instance: ``docker compose -f docker/docker-compose.yml up -d``.
|
|
2. Export env vars so the kit can discover the database::
|
|
|
|
export SQL_HOST=localhost
|
|
export SQL_PORT=5432
|
|
export SQL_USER=appuser
|
|
export SQL_PASSWORD=changeme
|
|
export SQL_DATABASE=appdb
|
|
export SQL_SSLMODE=disable
|
|
|
|
3. Install dependencies (``pip install sqlmodel-pg-kit[async]`` or ``uv pip install .``).
|
|
|
|
Run:
|
|
``python examples/07_postgres_minimal.py``
|
|
"""
|
|
|
|
from typing import Optional, List
|
|
|
|
from sqlalchemy import func
|
|
from sqlmodel import Field, SQLModel, select
|
|
|
|
from sqlmodel_pg_kit import Repository, create_all, get_session
|
|
|
|
|
|
class Widget(SQLModel, table=True):
|
|
id: Optional[int] = Field(default=None, primary_key=True)
|
|
name: str
|
|
in_stock: bool = True
|
|
|
|
|
|
if __name__ == "__main__":
|
|
create_all()
|
|
|
|
repo = Repository(Widget)
|
|
|
|
with get_session() as session:
|
|
# In web apps it's common to start each request by ensuring you have a clean view of data.
|
|
session.execute(Widget.__table__.delete())
|
|
session.commit()
|
|
|
|
# Create: typical POST handler persisting a single payload.
|
|
widget = repo.create(session, {"name": "rocket"})
|
|
print("Created:", widget)
|
|
|
|
# Bulk insert: useful for admin imports or seeding catalogs.
|
|
repo.bulk_insert(
|
|
session,
|
|
[
|
|
{"name": "satellite", "in_stock": True},
|
|
{"name": "capsule", "in_stock": False},
|
|
{"name": "probe", "in_stock": True},
|
|
],
|
|
)
|
|
|
|
# Read by id: GET /widgets/{id}
|
|
same = repo.get(session, widget.id)
|
|
print("Fetched by primary key:", same)
|
|
|
|
# Query a paginated listing ordered by newest first, backing a GET /widgets endpoint.
|
|
page: List[Widget] = repo.list(
|
|
session,
|
|
where=Widget.in_stock.is_(True),
|
|
order_by=[Widget.id.desc()],
|
|
page=1,
|
|
size=2,
|
|
)
|
|
print("First page of in-stock widgets:", [(w.id, w.name) for w in page])
|
|
|
|
# Free-form filtering with SQLModel/SQLAlchemy expressions for search endpoints.
|
|
search_term = "ro"
|
|
search_stmt = (
|
|
select(Widget)
|
|
.where(Widget.name.ilike(f"%{search_term}%"))
|
|
.order_by(Widget.name.asc())
|
|
)
|
|
matches = session.exec(search_stmt).all()
|
|
print(f"Search '{search_term}' ->", [(w.id, w.name) for w in matches])
|
|
|
|
# Aggregate counts to power dashboards or pagination metadata.
|
|
count_stmt = select(func.count(Widget.id)).where(Widget.in_stock.is_(True))
|
|
# scalar() keeps compatibility across SQLAlchemy versions
|
|
inventory_count = session.exec(count_stmt).scalar() or 0
|
|
print("In-stock count:", inventory_count)
|
|
|
|
# Update: PATCH /widgets/{id}
|
|
updated = repo.update(session, widget.id, in_stock=False)
|
|
print("Updated:", updated)
|
|
|
|
# Delete: DELETE /widgets/{id}
|
|
removed = repo.delete(session, widget.id)
|
|
print("Deleted:", removed)
|