File: builders.py

package info (click to toggle)
sqlalchemy-i18n 1.1.0-3
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 396 kB
  • sloc: python: 2,164; makefile: 157
file content (229 lines) | stat: -rw-r--r-- 7,974 bytes parent folder | download | duplicates (4)
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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
from copy import copy

import sqlalchemy as sa
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm.collections import attribute_mapped_collection
from sqlalchemy_utils.functions import get_primary_keys

from .comparators import TranslationComparator
from .exc import ImproperlyConfigured
from .expressions import current_locale
from .utils import get_fallback_locale, option


class HybridPropertyBuilder(object):
    def __init__(self, manager, translation_model):
        self.manager = manager
        self.translation_model = translation_model
        self.model = self.translation_model.__parent_class__

    def getter_factory(self, property_name):
        def attribute_getter(obj):
            value = getattr(obj.current_translation, property_name)
            if value:
                return value

            locale = get_fallback_locale(obj)
            return getattr(
                obj.translations[locale],
                property_name
            )

        return attribute_getter

    def setter_factory(self, property_name):
        """
        Return a hybrid property setter for given property name.

        :param property_name: Name of the property to generate a setter for
        """
        return (
            lambda obj, value:
            setattr(obj.current_translation, property_name, value)
        )

    def generate_hybrid(self, property_name):
        """
        Generate a SQLAlchemy hybrid property for given translation model
        property.

        :param property_name:
            Name of the translation model property to generate hybrid property
            for.
        """
        setattr(
            self.model,
            property_name,
            hybrid_property(
                fget=self.getter_factory(property_name),
                fset=self.setter_factory(property_name),
                expr=lambda cls: getattr(
                    cls.__translatable__['class'], property_name
                )
            )
        )

    def detect_collisions(self, property_name):
        """
        Detect possible naming collisions for given property name.

        :raises sqlalchemy_i18n.exc.ImproperlyConfigured: if the model already
            has a property with given name
        """
        mapper = sa.inspect(self.model)
        if mapper.has_property(property_name):
            raise ImproperlyConfigured(
                "Attribute name collision detected. Could not create "
                "hybrid property for translated attribute '%s'. "
                "An attribute with the same already exists in parent "
                "class '%s'." % (
                    property_name,
                    self.model.__name__
                )
            )

    def __call__(self):
        mapper = sa.orm.class_mapper(self.translation_model)
        for column in mapper.local_table.c:
            exclude = self.manager.option(
                self.model, 'exclude_hybrid_properties'
            )

            if column.key in exclude or column.primary_key:
                continue

            self.detect_collisions(column.key)
            self.generate_hybrid(column.key)


class RelationshipBuilder(object):
    def __init__(self, manager, translation_cls):
        self.manager = manager
        self.translation_cls = translation_cls
        self.parent_cls = self.translation_cls.__parent_class__

    @property
    def primary_key_conditions(self):
        conditions = []
        for key in get_primary_keys(self.parent_cls).keys():
            conditions.append(
                getattr(self.parent_cls, key) ==
                getattr(self.translation_cls, key)
            )
        return conditions

    def assign_single_translations(self):
        mapper = sa.orm.class_mapper(self.parent_cls)
        for locale in option(self.parent_cls, 'locales'):
            key = '_translation_%s' % locale
            if mapper.has_property(key):
                continue

            conditions = self.primary_key_conditions
            conditions.append(self.translation_cls.locale == locale)
            mapper.add_property(key, sa.orm.relationship(
                self.translation_cls,
                primaryjoin=sa.and_(*conditions),
                foreign_keys=list(
                    get_primary_keys(self.parent_cls).values()
                ),
                uselist=False,
                viewonly=True
            ))

    def assign_fallback_translation(self):
        """
        Assign the current translation relationship for translatable parent
        class.
        """
        mapper = sa.orm.class_mapper(self.parent_cls)
        if not mapper.has_property('_fallback_translation'):
            conditions = self.primary_key_conditions
            conditions.append(
                self.translation_cls.locale ==
                get_fallback_locale(self.parent_cls)
            )

            mapper.add_property('_fallback_translation', sa.orm.relationship(
                self.translation_cls,
                primaryjoin=sa.and_(*conditions),
                foreign_keys=list(
                    get_primary_keys(self.parent_cls).values()
                ),
                viewonly=True,
                uselist=False
            ))

    def assign_current_translation(self):
        """
        Assign the current translation relationship for translatable parent
        class.
        """
        mapper = sa.orm.class_mapper(self.parent_cls)
        if not mapper.has_property('_current_translation'):
            conditions = self.primary_key_conditions
            conditions.append(
                self.translation_cls.locale == current_locale()
            )

            mapper.add_property('_current_translation', sa.orm.relationship(
                self.translation_cls,
                primaryjoin=sa.and_(*conditions),
                foreign_keys=list(
                    get_primary_keys(self.parent_cls).values()
                ),
                viewonly=True,
                uselist=False
            ))

    def assign_translations(self):
        """
        Assigns translations relationship for translatable model. The assigned
        attribute is a relationship to all translation locales.
        """
        mapper = sa.orm.class_mapper(self.parent_cls)
        if not mapper.has_property('_translations'):
            mapper.add_property('_translations', sa.orm.relationship(
                self.translation_cls,
                **self.get_translations_relationship_args()
            ))

    def get_translations_relationship_args(self):
        foreign_keys = [
            getattr(self.translation_cls, column_key)
            for column_key in get_primary_keys(self.parent_cls).keys()
        ]

        relationship_args = copy(
            self.manager.option(
                self.parent_cls,
                'translations_relationship_args'
            )
        )
        defaults = dict(
            primaryjoin=sa.and_(*self.primary_key_conditions),
            foreign_keys=foreign_keys,
            collection_class=attribute_mapped_collection('locale'),
            comparator_factory=TranslationComparator,
            cascade='all, delete-orphan',
            passive_deletes=True,
        )
        for key, value in defaults.items():
            relationship_args.setdefault(key, value)
        return relationship_args

    def assign_translation_parent(self):
        mapper = sa.orm.class_mapper(self.translation_cls)
        if not mapper.has_property('translation_parent'):
            mapper.add_property('translation_parent', sa.orm.relationship(
                self.parent_cls,
                uselist=False,
                viewonly=True
            ))

    def __call__(self):
        self.assign_single_translations()
        self.assign_current_translation()
        self.assign_fallback_translation()
        self.assign_translations()
        self.assign_translation_parent()