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
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
|
import pytest
import sqlalchemy as sa
import sqlalchemy.orm
from sqlalchemy_utils import aggregated
@pytest.fixture
def Category(Base):
class Category(Base):
__tablename__ = 'category'
id = sa.Column(sa.Integer, primary_key=True)
name = sa.Column(sa.Unicode(255))
return Category
@pytest.fixture
def Catalog(Base, Category):
class Catalog(Base):
__tablename__ = 'catalog'
id = sa.Column(sa.Integer, primary_key=True)
name = sa.Column(sa.Unicode(255))
@aggregated(
'products.categories',
sa.Column(sa.Integer, default=0)
)
def category_count(self):
return sa.func.count(sa.distinct(Category.id))
return Catalog
@pytest.fixture
def Product(Base, Catalog, Category):
catalog_products = sa.Table(
'catalog_product',
Base.metadata,
sa.Column('catalog_id', sa.Integer, sa.ForeignKey('catalog.id')),
sa.Column('product_id', sa.Integer, sa.ForeignKey('product.id'))
)
product_categories = sa.Table(
'category_product',
Base.metadata,
sa.Column('category_id', sa.Integer, sa.ForeignKey('category.id')),
sa.Column('product_id', sa.Integer, sa.ForeignKey('product.id'))
)
class Product(Base):
__tablename__ = 'product'
id = sa.Column(sa.Integer, primary_key=True)
name = sa.Column(sa.Unicode(255))
price = sa.Column(sa.Numeric)
catalog_id = sa.Column(
sa.Integer, sa.ForeignKey('catalog.id')
)
catalogs = sa.orm.relationship(
Catalog,
backref='products',
secondary=catalog_products
)
categories = sa.orm.relationship(
Category,
backref='products',
secondary=product_categories
)
return Product
@pytest.fixture
def init_models(Category, Catalog, Product):
pass
@pytest.mark.usefixtures('postgresql_dsn')
class TestAggregateManyToManyAndManyToMany:
def test_insert(self, session, Product, Category, Catalog):
category = Category()
session.add(category)
products = [
Product(categories=[category]),
Product(categories=[category])
]
[session.add(product) for product in products]
catalog = Catalog(products=products)
session.add(catalog)
catalog2 = Catalog(products=products)
session.add(catalog2)
session.commit()
assert catalog.category_count == 1
assert catalog2.category_count == 1
|
