File: standalone_json.py

package info (click to toggle)
python-advanced-alchemy 1.4.1-1
  • links: PTS, VCS
  • area: main
  • in suites: sid, trixie
  • size: 3,708 kB
  • sloc: python: 25,811; makefile: 162; javascript: 123; sh: 4
file content (74 lines) | stat: -rw-r--r-- 2,299 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
# ruff: noqa: PLR2004, S101
from __future__ import annotations

import asyncio
from typing import TYPE_CHECKING, Any

from sqlalchemy.ext.asyncio import create_async_engine

from advanced_alchemy.base import UUIDBase
from advanced_alchemy.config import AsyncSessionConfig, SQLAlchemyAsyncConfig
from advanced_alchemy.repository import SQLAlchemyAsyncRepository

if TYPE_CHECKING:
    from sqlalchemy.orm import Mapped


class Item(UUIDBase):
    name: Mapped[str]
    # using ``Mapped[dict]`` with an AA provided base will map it to ``JSONB``
    data: Mapped[dict[str, Any]]


class ItemRepository(SQLAlchemyAsyncRepository[Item]):
    """Item repository."""

    model_type = Item


config = SQLAlchemyAsyncConfig(
    engine_instance=create_async_engine("postgresql+psycopg://app:super-secret@localhost:5432/app"),
    session_config=AsyncSessionConfig(expire_on_commit=False),
)


async def run_script() -> None:
    # Initializes the database.
    async with config.get_engine().begin() as conn:
        await conn.run_sync(Item.metadata.create_all)
    async with config.get_session() as db_session:
        repo = ItemRepository(session=db_session)
        # Add some data
        await repo.add_many(
            [
                Item(
                    name="Smartphone",
                    data={"price": 599.99, "brand": "XYZ"},
                ),
                Item(
                    name="Laptop",
                    data={"price": 1299.99, "brand": "ABC"},
                ),
                Item(
                    name="Headphones",
                    data={"not_price": 149.99, "brand": "DEF"},
                ),
            ],
            auto_commit=True,
        )

    async with config.get_session() as db_session:
        repo = ItemRepository(session=db_session)
        # Do some queries with JSON operations
        assert await repo.exists(Item.data["price"].as_float() == 599.99, Item.data["brand"].as_string() == "XYZ")

        assert await repo.count(Item.data.op("?")("price")) == 2

        products, total_products = await repo.list_and_count(Item.data.op("?")("not_price"))
        assert len(products) == 1
        assert total_products == 1
        assert products[0].name == "Headphones"


if __name__ == "__main__":
    asyncio.run(run_script())