6506453829

Committed 13 Oct 2023 09:24AM UTC coverage: 97.754% (+1.8%) from 95.959%

Build # 6506453829

Build Type

push

github

Committed by

indiVar0508

Commit Message

test: handle DB not being set for tests

Run Details

977 of 1066 branches covered (0.0%)

Branch coverage included in aggregate %.

2 of 2 new or added lines in 1 file covered. (100.0%)

15 existing lines in 1 file now uncovered.

4941 of 4988 relevant lines covered (99.06%)

9.75 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

83.68

/sqlalchemy_history/manager.py

"""Manager Module handles versioning configuration and communications between `builder` and `unit_of_work`
 package.

The Module contains following Class
- `VersioningManager`: VersioningManager delegates versioning configuration operations to builder classes
 and the actual versioning to UnitOfWork class.

"""
from functools import wraps

import sqlalchemy as sa
from sqlalchemy.orm import object_session
from sqlalchemy_utils import get_column_key

from sqlalchemy_history.builder import Builder
from sqlalchemy_history.fetcher import SubqueryFetcher, ValidityFetcher
from sqlalchemy_history.operation import Operation
from sqlalchemy_history.plugins import PluginCollection
from sqlalchemy_history.transaction import TransactionFactory
from sqlalchemy_history.unit_of_work import UnitOfWork
from sqlalchemy_history.utils import is_modified, is_versioned


def tracked_operation(func):
    @wraps(func)
    def wrapper(self, mapper, connection, target):
        if not is_versioned(target):
            return
        session = object_session(target)
        conn = session.connection()
        uow = self.get_uow(conn)
        return func(self, uow, target)

    return wrapper


class VersioningManager(object):
    """VersioningManager delegates versioning configuration operations to builder
    classes and the actual versioning to UnitOfWork class. Manager contains
    configuration options that act as defaults for all versioned classes.

    :param unit_of_work_cls: The UnitOfWork class to use for initializing UnitOfWork objects for
            versioning
    :param transaction_cls: Transaction class to use for versioning. If None, the default
            Transaction class generated by TransactionFactory will be used.
    :param user_cls: User class which Transaction class should have relationship to. This
            can either be a class or string name of a class for lazy evaluation.
    :param options: Versioning options
    :param plugins: Versioning plugins that listen the events invoked by the manager.
    :param builder: Builder object which handles the building of versioning tables and
            models.

    """

    def __init__(
        self,
        unit_of_work_cls=UnitOfWork,
        transaction_cls=None,
        user_cls=None,
        options={},
        plugins=None,
        builder=None,
    ):
        self.uow_class = unit_of_work_cls
        if builder is None:
            self.builder = Builder()
        else:
            self.builder = builder
        self.builder.manager = self
        self.reset()
        if transaction_cls is not None:
            self.transaction_cls = transaction_cls
        else:
            self.transaction_cls = TransactionFactory()
        if user_cls is not None:
            self.user_cls = user_cls

        self.options = {
            "versioning": True,
            "base_classes": None,
            "table_name": "%s_version",
            "exclude": [],
            "include": [],
            "create_models": True,
            "create_tables": True,
            "transaction_column_name": "transaction_id",
            "end_transaction_column_name": "end_transaction_id",
            "operation_type_column_name": "operation_type",
            "strategy": "validity",
            "use_module_name": False,
        }
        if plugins is None:
            self.plugins = []
        else:
            self.plugins = plugins
        self.options.update(options)

    @property
    def plugins(self):
        return self._plugins

    @plugins.setter
    def plugins(self, plugin_collection):
        self._plugins = PluginCollection(plugin_collection)

    def fetcher(self, obj):
        if self.option(obj, "strategy") == "subquery":
            return SubqueryFetcher(self)
        else:
            return ValidityFetcher(self)

    def get_uow(self, conn):
        try:
            return self.units_of_work[conn]
        except KeyError:
            try:
                return self.units_of_work[conn.engine]
            except KeyError:
                for connection in self.units_of_work.keys():
                    if not connection.closed and connection.connection is conn.connection:
                        return self.unit_of_work(conn.session)
                else:
                    raise

    def reset(self):
        """Resets this manager's internal state.

        This method should be used in test cases that create models on the fly.
        Otherwise history_class_map and some other variables would be polluted
        by no more used model classes.


        """
        self.tables = {}
        self.pending_classes = []
        self.version_table_map = {}  # Key is the parent table, Value is the version table
        self.declarative_base = None
        self.version_class_map = {}  # Key is the parent model, Value is the version model
        self.session_listeners = {
            "before_flush": self.before_flush,
            "after_flush": self.after_flush,
            "after_commit": self.clear,
            "after_rollback": self.clear,
        }
        self.mapper_listeners = {
            "after_delete": self.track_deletes,
            "after_update": self.track_updates,
            "after_insert": self.track_inserts,
        }
        self.class_config_listeners = {
            "instrument_class": self.builder.instrument_versioned_classes,
            "after_configured": self.builder.configure_versioned_classes,
        }

        # A dictionary of units of work. Keys as connection objects and values
        # as UnitOfWork objects.
        self.units_of_work = {}

        self.session_connection_map = {}

        self.metadata = None

    def create_transaction_model(self):
        """Create Transaction class but only if it doesn't already exist in declarative model registry."""
        if isinstance(self.transaction_cls, TransactionFactory):
            self.transaction_cls = self.transaction_cls(self)
        return self.transaction_cls

    def is_excluded_column(self, model, column):
        try:
            key = get_column_key(model, column)
        except sa.orm.exc.UnmappedColumnError:
            return False

        return self.is_excluded_property(model, key)

    def is_excluded_property(self, model, key):
        """Returns whether or not given property of given model is excluded from the associated history model.

        :param model: SQLAlchemy declarative model object.
        :param key: Model property key

        """
        if key in self.option(model, "include"):
            return False
        return key in self.option(model, "exclude")

    def option(self, model, name):
        """Returns the option value for given model.

        If the option is not found from given model falls back to default values of this manager object.
        If the option is not found from this manager object either this method throws a KeyError.

        :param model: SQLAlchemy declarative object
        :param name: name of the versioning option

        """
        if not hasattr(model, "__versioned__"):
            raise TypeError("Model %r is not versioned." % model)
        try:
            return model.__versioned__[name]
        except KeyError:
            return self.options[name]

    def apply_class_configuration_listeners(self, mapper):
        """Applies class configuration listeners for given mapper.

        The listener work in two phases:

        1. Class instrumentation phase
            The first listeners listens to class instrumentation event and
            handles the collecting of versioned models and adds them to
            the pending_classes list.
        2. After class configuration phase
            The second listener listens to after class configuration event and
            handles the actual history model generation based on list that
            was collected during class instrumenation phase.

        :param mapper: SQLAlchemy mapper to apply the class configuration listeners to

        """
        for event_name, listener in self.class_config_listeners.items():
            sa.event.listen(mapper, event_name, listener)

    def remove_class_configuration_listeners(self, mapper):
        """Remove versioning class configuration listeners from specified mapper.

        :param mapper: mapper to remove class configuration listeners from

        """
        for event_name, listener in self.class_config_listeners.items():
            sa.event.remove(mapper, event_name, listener)

    def track_operations(self, mapper):
        """Attach listeners for specified mapper that track SQL inserts, updates and deletes.

        :param mapper: mapper to track the SQL operations from

        """
        for event_name, listener in self.mapper_listeners.items():
            sa.event.listen(mapper, event_name, listener)

    def remove_operations_tracking(self, mapper):
        """Remove listeners from specified mapper that track SQL inserts, updates and deletes.

        :param mapper: mapper to remove the SQL operations tracking listeners from

        """
        for event_name, listener in self.mapper_listeners.items():
            sa.event.remove(mapper, event_name, listener)

    def track_session(self, session):
        """Attach listeners that track the operations (flushing, committing and
        rolling back) of given session.

        This method should be used in conjunction with `track_operations`.

        :param session: SQLAlchemy session to track the operations from

        """
        for event_name, listener in self.session_listeners.items():
            sa.event.listen(session, event_name, listener)

    def remove_session_tracking(self, session):
        """Remove listeners that track the operations (flushing, committing and rolling back) of given
         session.
        This method should be used in conjunction with `remove_operations_tracking`.

        :param session: SQLAlchemy session to remove the operations tracking from

        """
        for event_name, listener in self.session_listeners.items():
            sa.event.remove(session, event_name, listener)

    @tracked_operation
    def track_inserts(self, uow, target):
        """Track object insert operations.

        Whenever object is inserted it is added to this UnitOfWork's internal operations dictionary.
        """
        uow.operations.add_insert(target)

    @tracked_operation
    def track_updates(self, uow, target):
        """Track object update operations.

        Whenever object is updated it is added to this UnitOfWork's internal operations dictionary.
        """
        if not is_modified(target):
            return
        uow.operations.add_update(target)

    @tracked_operation
    def track_deletes(self, uow, target):
        """Track object deletion operations.
        Whenever object is deleted it is added to this UnitOfWork's internal operations dictionary.
        """
        uow.operations.add_delete(target)

    def unit_of_work(self, session):
        """Return the associated SQLAlchemy-History UnitOfWork object for given SQLAlchemy session object.

        If no UnitOfWork object exists for given object then this method tries to create one.

        :param session: SQLAlchemy session object
        """
        conn = session.connection()
        if conn not in self.session_connection_map.values():
            self.session_connection_map[session] = conn

        if conn in self.units_of_work:
            return self.units_of_work[conn]
        else:
            uow = self.uow_class(self)
            self.units_of_work[conn] = uow
            return uow

    def before_flush(self, session, flush_context, instances):
        """Before flush listener for SQLAlchemy sessions.
        If this manager has versioning enabled this listener invokes the process before flush of associated
         UnitOfWork object.

        :param session: SQLAlchemy session
        :param flush_context:
        :param instances:

        """
        if not self.options["versioning"]:
            return

        uow = self.unit_of_work(session)
        uow.process_before_flush(session)

    def after_flush(self, session, flush_context):
        """After flush listener for SQLAlchemy sessions.

        If this manager has versioning enabled this listener gets the UnitOfWork associated with
        session's connections and invokes the process_after_flush method of that object.

        :param session: SQLAlchemy session
        :param flush_context:

        """
        if not self.options["versioning"]:
            return
        uow = self.unit_of_work(session)
        uow.process_after_flush(session)

    def clear(self, session):
        """Simple SQLAlchemy listener that is being invoked after successful transaction commit or when
         transaction rollback occurs.
        The purpose of this listener is to reset this UnitOfWork back to its initialization state.

        :param session: SQLAlchemy session object

        """
        if session.transaction.nested:
            return
        conn = self.session_connection_map.pop(session, None)
        if conn is None:
            return

        if conn in self.units_of_work:
            uow = self.units_of_work[conn]
            uow.reset(session)
            del self.units_of_work[conn]

        for connection in dict(self.units_of_work).keys():
            if connection.closed or conn.connection is connection.connection:
                uow = self.units_of_work[connection]
                uow.reset(session)
                del self.units_of_work[connection]

    def clear_connection(self, conn):
        if conn in self.units_of_work:
            uow = self.units_of_work[conn]
            uow.reset()
            del self.units_of_work[conn]

        for session, connection in dict(self.session_connection_map).items():
            if connection is conn:
                del self.session_connection_map[session]

        for connection in dict(self.units_of_work).keys():
            if connection.closed or conn.connection is connection.connection:
                uow = self.units_of_work[connection]
                uow.reset()
                del self.units_of_work[connection]

    def append_association_operation(self, conn, table_name, params, op):
        """Append history association operation to pending_statements list.

        :param conn:
        :param table_name:
        :param params:
        :param op:

        """
        stmt = (
            self.metadata.tables[self.options["table_name"] % table_name]
            .insert()
            .values({**params, "operation_type": op})
        )
        uow = self.get_uow(conn)
        uow.pending_statements.append(stmt)

    def track_cloned_connections(self, c, opt):
        """Track cloned connections from association tables.

        :param c:
        :param opt:

        """
        if c not in self.units_of_work.keys():
            for connection, uow in dict(self.units_of_work).items():
                if (
                    not connection.closed and connection.connection is c.connection
                ):  # ConnectionFairy is the same - this is a clone
                    self.units_of_work[c] = uow

    def track_sql_operations(self, conn, cursor, statement, parameters, context, executemany):
        """
        This function tracks all SQLoperators directly done by the sqlalchemy cursor.
        We use it to track operations on tables which are not mapped to a ORM model.

        This is mainly used to generate the history of association tables by adding
        those operations to pending_statements list.
        If we later want to track datatables also we may need to handle autoincrement and Edit operations
        FOr now, assoc tables will never have autoincrement nor do they modify an existing record

        :param conn:
        :param cursor:
        :param statement:
        :param parameters:
        :param context:
        :param executemany:

        """
        if not self.options["versioning"]:
            return

        op = None
        if context.isinsert:
            op = Operation.INSERT
        elif context.isdelete:
            op = Operation.DELETE

        if op is not None:
            table_name = (
                context.invoked_statement.table.schema + "." + context.invoked_statement.table.name
                if context.invoked_statement.table.schema
                else context.invoked_statement.table.name
            )
            # ORM tables are tracked using `mapper_listeners`. We only need to track table which do not
            #  have mappers here
            orm_tracked_tables = {
                c.__table__
                for c in self.version_class_map
                # NOTE: We add hasattr(c, '__table__') cause some ORM may not have a physical table
                #  associated to them
                if hasattr(c, "__table__")
            }
            table_names = [
                table.name if not table.schema else table.schema + "." + table.name
                for table in self.version_table_map
                if table not in orm_tracked_tables
            ]
            if table_name in table_names:
                for params in context.compiled_parameters:
                    self.append_association_operation(conn, table_name, params, op)

1	"""Manager Module handles versioning configuration and communications between `builder` and `unit_of_work`	10✔
2	package.
3
4	The Module contains following Class
5	- `VersioningManager`: VersioningManager delegates versioning configuration operations to builder classes
6	and the actual versioning to UnitOfWork class.
7
8	"""
9	from functools import wraps	10✔
10
11	import sqlalchemy as sa	10✔
12	from sqlalchemy.orm import object_session	10✔
13	from sqlalchemy_utils import get_column_key	10✔
14
15	from sqlalchemy_history.builder import Builder	10✔
16	from sqlalchemy_history.fetcher import SubqueryFetcher, ValidityFetcher	10✔
17	from sqlalchemy_history.operation import Operation	10✔
18	from sqlalchemy_history.plugins import PluginCollection	10✔
19	from sqlalchemy_history.transaction import TransactionFactory	10✔
20	from sqlalchemy_history.unit_of_work import UnitOfWork	10✔
21	from sqlalchemy_history.utils import is_modified, is_versioned	10✔
22
23
24	def tracked_operation(func):	10✔
25	@wraps(func)	10✔
26	def wrapper(self, mapper, connection, target):	5✔
27	if not is_versioned(target):	10✔
28	return	10✔
29	session = object_session(target)	10✔
30	conn = session.connection()	10✔
31	uow = self.get_uow(conn)	10✔
32	return func(self, uow, target)	10✔
33
34	return wrapper	10✔
35
36
37	class VersioningManager(object):	10✔
38	"""VersioningManager delegates versioning configuration operations to builder	10✔
39	classes and the actual versioning to UnitOfWork class. Manager contains
40	configuration options that act as defaults for all versioned classes.
41
42	:param unit_of_work_cls: The UnitOfWork class to use for initializing UnitOfWork objects for
43	versioning
44	:param transaction_cls: Transaction class to use for versioning. If None, the default
45	Transaction class generated by TransactionFactory will be used.
46	:param user_cls: User class which Transaction class should have relationship to. This
47	can either be a class or string name of a class for lazy evaluation.
48	:param options: Versioning options
49	:param plugins: Versioning plugins that listen the events invoked by the manager.
50	:param builder: Builder object which handles the building of versioning tables and
51	models.
52
53	"""
54
55	def __init__(	10✔
56	self,
57	unit_of_work_cls=UnitOfWork,
58	transaction_cls=None,
59	user_cls=None,
60	options={},
61	plugins=None,
62	builder=None,
63	):
64	self.uow_class = unit_of_work_cls	10✔
65	if builder is None:	10✔
66	self.builder = Builder()	10✔
67	else:
UNCOV 68	self.builder = builder	×
69	self.builder.manager = self	10✔
70	self.reset()	10✔
71	if transaction_cls is not None:	10✔
UNCOV 72	self.transaction_cls = transaction_cls	×
73	else:
74	self.transaction_cls = TransactionFactory()	10✔
75	if user_cls is not None:	10✔
UNCOV 76	self.user_cls = user_cls	×
77
78	self.options = {	10✔
79	"versioning": True,
80	"base_classes": None,
81	"table_name": "%s_version",
82	"exclude": [],
83	"include": [],
84	"create_models": True,
85	"create_tables": True,
86	"transaction_column_name": "transaction_id",
87	"end_transaction_column_name": "end_transaction_id",
88	"operation_type_column_name": "operation_type",
89	"strategy": "validity",
90	"use_module_name": False,
91	}
92	if plugins is None:	10✔
93	self.plugins = []	10✔
94	else:
UNCOV 95	self.plugins = plugins	×
96	self.options.update(options)	10✔
97
98	@property	10✔
99	def plugins(self):	5✔
100	return self._plugins	10✔
101
102	@plugins.setter	10✔
103	def plugins(self, plugin_collection):	5✔
104	self._plugins = PluginCollection(plugin_collection)	10✔
105
106	def fetcher(self, obj):	10✔
107	if self.option(obj, "strategy") == "subquery":	10✔
108	return SubqueryFetcher(self)	10✔
109	else:
110	return ValidityFetcher(self)	10✔
111
112	def get_uow(self, conn):	10✔
113	try:	10✔
114	return self.units_of_work[conn]	10✔
115	except KeyError:	10✔
116	try:	10✔
117	return self.units_of_work[conn.engine]	10✔
118	except KeyError:	10✔
119	for connection in self.units_of_work.keys():	10!
UNCOV 120	if not connection.closed and connection.connection is conn.connection:	×
UNCOV 121	return self.unit_of_work(conn.session)	×
122	else:
123	raise	10✔
124
125	def reset(self):	10✔
126	"""Resets this manager's internal state.
127
128	This method should be used in test cases that create models on the fly.
129	Otherwise history_class_map and some other variables would be polluted
130	by no more used model classes.
131
132
133	"""
134	self.tables = {}	10✔
135	self.pending_classes = []	10✔
136	self.version_table_map = {} # Key is the parent table, Value is the version table	10✔
137	self.declarative_base = None	10✔
138	self.version_class_map = {} # Key is the parent model, Value is the version model	10✔
139	self.session_listeners = {	10✔
140	"before_flush": self.before_flush,
141	"after_flush": self.after_flush,
142	"after_commit": self.clear,
143	"after_rollback": self.clear,
144	}
145	self.mapper_listeners = {	10✔
146	"after_delete": self.track_deletes,
147	"after_update": self.track_updates,
148	"after_insert": self.track_inserts,
149	}
150	self.class_config_listeners = {	10✔
151	"instrument_class": self.builder.instrument_versioned_classes,
152	"after_configured": self.builder.configure_versioned_classes,
153	}
154
155	# A dictionary of units of work. Keys as connection objects and values
156	# as UnitOfWork objects.
157	self.units_of_work = {}	10✔
158
159	self.session_connection_map = {}	10✔
160
161	self.metadata = None	10✔
162
163	def create_transaction_model(self):	10✔
164	"""Create Transaction class but only if it doesn't already exist in declarative model registry."""
165	if isinstance(self.transaction_cls, TransactionFactory):	10✔
166	self.transaction_cls = self.transaction_cls(self)	10✔
167	return self.transaction_cls	10!
168
169	def is_excluded_column(self, model, column):	10✔
170	try:	10✔
171	key = get_column_key(model, column)	10✔
172	except sa.orm.exc.UnmappedColumnError:	10✔
173	return False	10✔
174
175	return self.is_excluded_property(model, key)	10✔
176
177	def is_excluded_property(self, model, key):	10✔
178	"""Returns whether or not given property of given model is excluded from the associated history model.
179
180	:param model: SQLAlchemy declarative model object.
181	:param key: Model property key
182
183	"""
184	if key in self.option(model, "include"):	10✔
185	return False	10✔
186	return key in self.option(model, "exclude")	10✔
187
188	def option(self, model, name):	10✔
189	"""Returns the option value for given model.
190
191	If the option is not found from given model falls back to default values of this manager object.
192	If the option is not found from this manager object either this method throws a KeyError.
193
194	:param model: SQLAlchemy declarative object
195	:param name: name of the versioning option
196
197	"""
198	if not hasattr(model, "__versioned__"):	10✔
199	raise TypeError("Model %r is not versioned." % model)	10✔
200	try:	10✔
201	return model.__versioned__[name]	10✔
202	except KeyError:	10✔
203	return self.options[name]	10✔
204
205	def apply_class_configuration_listeners(self, mapper):	10✔
206	"""Applies class configuration listeners for given mapper.
207
208	The listener work in two phases:
209
210	1. Class instrumentation phase
211	The first listeners listens to class instrumentation event and
212	handles the collecting of versioned models and adds them to
213	the pending_classes list.
214	2. After class configuration phase
215	The second listener listens to after class configuration event and
216	handles the actual history model generation based on list that
217	was collected during class instrumenation phase.
218
219	:param mapper: SQLAlchemy mapper to apply the class configuration listeners to
220
221	"""
222	for event_name, listener in self.class_config_listeners.items():	10✔
223	sa.event.listen(mapper, event_name, listener)	10✔
224
225	def remove_class_configuration_listeners(self, mapper):	10✔
226	"""Remove versioning class configuration listeners from specified mapper.
227
228	:param mapper: mapper to remove class configuration listeners from
229
230	"""
231	for event_name, listener in self.class_config_listeners.items():	10✔
232	sa.event.remove(mapper, event_name, listener)	10✔
233
234	def track_operations(self, mapper):	10✔
235	"""Attach listeners for specified mapper that track SQL inserts, updates and deletes.
236
237	:param mapper: mapper to track the SQL operations from
238
239	"""
240	for event_name, listener in self.mapper_listeners.items():	10✔
241	sa.event.listen(mapper, event_name, listener)	10✔
242
243	def remove_operations_tracking(self, mapper):	10✔
244	"""Remove listeners from specified mapper that track SQL inserts, updates and deletes.
245
246	:param mapper: mapper to remove the SQL operations tracking listeners from
247
248	"""
249	for event_name, listener in self.mapper_listeners.items():	10✔
250	sa.event.remove(mapper, event_name, listener)	10✔
251
252	def track_session(self, session):	10✔
253	"""Attach listeners that track the operations (flushing, committing and
254	rolling back) of given session.
255
256	This method should be used in conjunction with `track_operations`.
257
258	:param session: SQLAlchemy session to track the operations from
259
260	"""
261	for event_name, listener in self.session_listeners.items():	10✔
262	sa.event.listen(session, event_name, listener)	10✔
263
264	def remove_session_tracking(self, session):	10✔
265	"""Remove listeners that track the operations (flushing, committing and rolling back) of given
266	session.
267	This method should be used in conjunction with `remove_operations_tracking`.
268
269	:param session: SQLAlchemy session to remove the operations tracking from
270
271	"""
272	for event_name, listener in self.session_listeners.items():	10✔
273	sa.event.remove(session, event_name, listener)	10✔
274
275	@tracked_operation	10✔
276	def track_inserts(self, uow, target):	5✔
277	"""Track object insert operations.
278
279	Whenever object is inserted it is added to this UnitOfWork's internal operations dictionary.
280	"""
281	uow.operations.add_insert(target)	10✔
282
283	@tracked_operation	10✔
284	def track_updates(self, uow, target):	5✔
285	"""Track object update operations.
286
287	Whenever object is updated it is added to this UnitOfWork's internal operations dictionary.
288	"""
289	if not is_modified(target):	10✔
290	return	10✔
291	uow.operations.add_update(target)	10✔
292
293	@tracked_operation	10✔
294	def track_deletes(self, uow, target):	5✔
295	"""Track object deletion operations.
296	Whenever object is deleted it is added to this UnitOfWork's internal operations dictionary.
297	"""
298	uow.operations.add_delete(target)	10✔
299
300	def unit_of_work(self, session):	10✔
301	"""Return the associated SQLAlchemy-History UnitOfWork object for given SQLAlchemy session object.
302
303	If no UnitOfWork object exists for given object then this method tries to create one.
304
305	:param session: SQLAlchemy session object
306	"""
307	conn = session.connection()	10✔
308	if conn not in self.session_connection_map.values():	10✔
309	self.session_connection_map[session] = conn	10✔
310
311	if conn in self.units_of_work:	10✔
312	return self.units_of_work[conn]	10✔
313	else:
314	uow = self.uow_class(self)	10✔
315	self.units_of_work[conn] = uow	10✔
316	return uow	10✔
317
318	def before_flush(self, session, flush_context, instances):	10✔
319	"""Before flush listener for SQLAlchemy sessions.
320	If this manager has versioning enabled this listener invokes the process before flush of associated
321	UnitOfWork object.
322
323	:param session: SQLAlchemy session
324	:param flush_context:
325	:param instances:
326
327	"""
328	if not self.options["versioning"]:	10✔
UNCOV 329	return	×
330
331	uow = self.unit_of_work(session)	10✔
332	uow.process_before_flush(session)	10✔
333
334	def after_flush(self, session, flush_context):	10✔
335	"""After flush listener for SQLAlchemy sessions.
336
337	If this manager has versioning enabled this listener gets the UnitOfWork associated with
338	session's connections and invokes the process_after_flush method of that object.
339
340	:param session: SQLAlchemy session
341	:param flush_context:
342
343	"""
344	if not self.options["versioning"]:	10✔
UNCOV 345	return	×
346	uow = self.unit_of_work(session)	10✔
347	uow.process_after_flush(session)	10✔
348
349	def clear(self, session):	10✔
350	"""Simple SQLAlchemy listener that is being invoked after successful transaction commit or when
351	transaction rollback occurs.
352	The purpose of this listener is to reset this UnitOfWork back to its initialization state.
353
354	:param session: SQLAlchemy session object
355
356	"""
357	if session.transaction.nested:	10✔
358	return	10✔
359	conn = self.session_connection_map.pop(session, None)	10✔
360	if conn is None:	10✔
361	return	10✔
362
363	if conn in self.units_of_work:	10✔
364	uow = self.units_of_work[conn]	10✔
365	uow.reset(session)	10✔
366	del self.units_of_work[conn]	10✔
367
368	for connection in dict(self.units_of_work).keys():	10!
369	if connection.closed or conn.connection is connection.connection:	10✔
370	uow = self.units_of_work[connection]	2!
371	uow.reset(session)	2✔
372	del self.units_of_work[connection]	2✔
373
374	def clear_connection(self, conn):	10✔
375	if conn in self.units_of_work:	10✔
376	uow = self.units_of_work[conn]	10✔
377	uow.reset()	10✔
378	del self.units_of_work[conn]	10✔
379
380	for session, connection in dict(self.session_connection_map).items():	10!
381	if connection is conn:	10✔
382	del self.session_connection_map[session]	10✔
383
384	for connection in dict(self.units_of_work).keys():	10!
385	if connection.closed or conn.connection is connection.connection:	×
386	uow = self.units_of_work[connection]	×
UNCOV 387	uow.reset()	×
UNCOV 388	del self.units_of_work[connection]	×
389
390	def append_association_operation(self, conn, table_name, params, op):	10✔
391	"""Append history association operation to pending_statements list.
392
393	:param conn:
394	:param table_name:
395	:param params:
396	:param op:
397
398	"""
399	stmt = (	10✔
400	self.metadata.tables[self.options["table_name"] % table_name]
401	.insert()
402	.values({**params, "operation_type": op})
403	)
404	uow = self.get_uow(conn)	10✔
405	uow.pending_statements.append(stmt)	10✔
406
407	def track_cloned_connections(self, c, opt):	10✔
408	"""Track cloned connections from association tables.
409
410	:param c:
411	:param opt:
412
413	"""
UNCOV 414	if c not in self.units_of_work.keys():	×
UNCOV 415	for connection, uow in dict(self.units_of_work).items():	×
UNCOV 416	if (	×
417	not connection.closed and connection.connection is c.connection
418	): # ConnectionFairy is the same - this is a clone
UNCOV 419	self.units_of_work[c] = uow	×
420
421	def track_sql_operations(self, conn, cursor, statement, parameters, context, executemany):	10✔
422	"""
423	This function tracks all SQLoperators directly done by the sqlalchemy cursor.
424	We use it to track operations on tables which are not mapped to a ORM model.
425
426	This is mainly used to generate the history of association tables by adding
427	those operations to pending_statements list.
428	If we later want to track datatables also we may need to handle autoincrement and Edit operations
429	FOr now, assoc tables will never have autoincrement nor do they modify an existing record
430
431	:param conn:
432	:param cursor:
433	:param statement:
434	:param parameters:
435	:param context:
436	:param executemany:
437
438	"""
439	if not self.options["versioning"]:	10✔
UNCOV 440	return	×
441
442	op = None	10✔
443	if context.isinsert:	10✔
444	op = Operation.INSERT	10✔
445	elif context.isdelete:	10✔
446	op = Operation.DELETE	10✔
447
448	if op is not None:	10✔
449	table_name = (	10✔
450	context.invoked_statement.table.schema + "." + context.invoked_statement.table.name
451	if context.invoked_statement.table.schema
452	else context.invoked_statement.table.name
453	)
454	# ORM tables are tracked using `mapper_listeners`. We only need to track table which do not
455	# have mappers here
456	orm_tracked_tables = {	10✔
457	c.__table__
458	for c in self.version_class_map
459	# NOTE: We add hasattr(c, '__table__') cause some ORM may not have a physical table
460	# associated to them
461	if hasattr(c, "__table__")
462	}
463	table_names = [	10✔
464	table.name if not table.schema else table.schema + "." + table.name
465	for table in self.version_table_map
466	if table not in orm_tracked_tables
467	]
468	if table_name in table_names:	10✔
469	for params in context.compiled_parameters:	10✔
470	self.append_association_operation(conn, table_name, params, op)	10✔

corridor / sqlalchemy-history / 6506453829

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous