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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
|
import pytest
import sqlalchemy as sa
import sqlalchemy.orm
from sqlalchemy_utils import aggregated
@pytest.fixture
def Catalog(Base):
class Catalog(Base):
__tablename__ = 'catalog'
id = sa.Column(sa.Integer, primary_key=True)
@aggregated(
'categories.sub_categories.products',
sa.Column(sa.Integer, default=0)
)
def product_count(self):
return sa.func.count('1')
categories = sa.orm.relationship('Category', backref='catalog')
return Catalog
@pytest.fixture
def Category(Base):
class Category(Base):
__tablename__ = 'category'
id = sa.Column(sa.Integer, primary_key=True)
catalog_id = sa.Column(sa.Integer, sa.ForeignKey('catalog.id'))
sub_categories = sa.orm.relationship(
'SubCategory', backref='category'
)
return Category
@pytest.fixture
def SubCategory(Base):
class SubCategory(Base):
__tablename__ = 'sub_category'
id = sa.Column(sa.Integer, primary_key=True)
category_id = sa.Column(sa.Integer, sa.ForeignKey('category.id'))
products = sa.orm.relationship('Product', backref='sub_category')
return SubCategory
@pytest.fixture
def Product(Base):
class Product(Base):
__tablename__ = 'product'
id = sa.Column(sa.Integer, primary_key=True)
price = sa.Column(sa.Numeric)
sub_category_id = sa.Column(
sa.Integer, sa.ForeignKey('sub_category.id')
)
return Product
@pytest.fixture
def init_models(Catalog, Category, SubCategory, Product):
pass
@pytest.fixture
def catalog_factory(Product, SubCategory, Category, Catalog, session):
def catalog_factory():
product = Product()
session.add(product)
sub_category = SubCategory(products=[product])
session.add(sub_category)
category = Category(sub_categories=[sub_category])
session.add(category)
catalog = Catalog(categories=[category])
session.add(catalog)
return catalog
return catalog_factory
@pytest.mark.usefixtures('postgresql_dsn')
class Test3LevelDeepOneToMany:
def test_assigns_aggregates(self, session, catalog_factory):
catalog = catalog_factory()
session.commit()
session.refresh(catalog)
assert catalog.product_count == 1
def catalog_factory(
self,
session,
Product,
SubCategory,
Category,
Catalog
):
product = Product()
session.add(product)
sub_category = SubCategory(
products=[product]
)
session.add(sub_category)
category = Category(sub_categories=[sub_category])
session.add(category)
catalog = Catalog(categories=[category])
session.add(catalog)
return catalog
def test_only_updates_affected_aggregates(
self,
session,
catalog_factory,
Product
):
catalog = catalog_factory()
catalog2 = catalog_factory()
session.commit()
# force set catalog2 product_count to zero in order to check if it gets
# updated when the other catalog's product count gets updated
session.execute(
sa.text(
'UPDATE catalog SET product_count = 0 WHERE id = %d' % catalog2.id
)
)
catalog.categories[0].sub_categories[0].products.append(
Product()
)
session.commit()
session.refresh(catalog)
session.refresh(catalog2)
assert catalog.product_count == 2
assert catalog2.product_count == 0
|
