A => .hgignore +5 -0
@@ 0,0 1,5 @@
+syntax: rootglob
+
+.mypy_cache/
+.tox/
+*.egg-info/
A => LICENSE.txt +202 -0
@@ 0,0 1,202 @@
+
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+ 1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+ 2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+ 3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+ 4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+ 5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+ 6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+ 7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+ 8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+ 9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+ END OF TERMS AND CONDITIONS
+
+ APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+ Copyright [yyyy] [name of copyright owner]
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
A => README.rst +2 -0
@@ 0,0 1,2 @@
+SQLFixture
+==========
A => pyproject.toml +15 -0
@@ 0,0 1,15 @@
+[build-system]
+requires = ["setuptools", "wheel"]
+
+[project]
+name = "sqlfixtures"
+version = "0.0.1"
+authors = [{name="Oliver Cope", email="oliver@redgecko.org"}]
+description = "Create test fixtures against an SQL database"
+readme = "README.rst"
+requires-python = ">=3.10"
+classifiers = [
+ "Programming Language :: Python :: 3",
+ "License :: OSI Approved :: MIT License",
+ "Operating System :: OS Independent",
+]
A => sqlfixtures/__init__.py +4 -0
@@ 0,0 1,4 @@
+from .insertable import Insertable
+from .sqlfixture import SQLFixture
+
+__all__ = ["Insertable", "SQLFixture"]
A => sqlfixtures/insertable.py +355 -0
@@ 0,0 1,355 @@
+import inspect
+import contextlib
+
+from typing import Any
+from typing import Iterable
+from itertools import chain
+
+from sqlfixtures.sqlfixture import SQLFixture
+
+
+class Insertable:
+ """
+
+ Example usage::
+
+
+ class sqlfixture:
+ user = Insertable(
+ "users",
+ username="user1@example.com",
+ email="user1@example.com",
+ )
+ advertiser = applicant(
+ username="advertiser@example.com", email="advertiser@example.com"
+ )
+ advert = Insertable(
+ "adverts",
+ origin=Insertable("origins", source="skot", owner_id=advertiser.id, owner=advertiser),
+ origin_id=Insertable.self().origin.id,
+ origin__owner=advertiser
+ )
+
+ with Insertable.insert(conn, sqlfixture) as f:
+ ...
+
+
+ Notes:
+
+ ``f`` will be returned as an instance of ``sqlfixture`` with the
+ attributes set to dataclass objects containg the inserted rows.
+
+ Insertables can reference other insertables::
+
+ class sqlfixture:
+ user = Insertable("users", name="bernice")
+ post = Insertable("posts", content="...", user_id=user.id)
+
+ Use Insertable.self() for access to Insertables created on the same
+ object::
+
+ class sqlfixture:
+ post = Insertable(
+ "posts",
+ content="...",
+ owner=Insertable("users", name="bernice"),
+ owner_id=Insertable.self().owner.id
+ )
+
+ Bear in mind that Insertable isn't part of an ORM and doesn't know
+ about database relationships! You **must** populate all foreign keys.
+ This won't work::
+
+ class sqlfixture:
+
+ user = Insertable("users", name="bernice")
+
+ # ERROR: doesn't set owner_id so the post will not be correctly
+ # inserted
+ post = Insertable("posts", content="...", owner=user)
+
+ """
+ _items: dict[str, Any]
+ _attributes: list[tuple[str, str]]
+ _lookup: dict[type, str] = {}
+ table: str
+
+ def __init__(self, *args: Any, **kwargs: Any):
+ self.__dict__["_items"] = {}
+ self.__dict__["_attributes"] = []
+ match args:
+ case (str(s),):
+ self.__dict__["table"], what = s, {}
+ case (str(t), w):
+ self.__dict__["table"], what = t, w
+ case (x,):
+ what = x
+ self.__dict__["table"] = self.__class__._lookup[type(what)]
+
+ match what:
+ case None:
+ pass
+ case dict(mapping):
+ for k, v in mapping.items():
+ self[k] = v
+ case _:
+ for k, v in inspect.getmembers(what):
+ self[k] = v
+ for k, v in kwargs.items():
+ if isinstance(v, Reference) and v._source is Reference.SELF:
+ self[k] = v.unwrap(self)
+ else:
+ self[k] = v
+
+ def __repr__(self):
+ return f"<{self.__class__.__name__} {self.table} {repr(self._items)[:40]}...>"
+
+ def __getattr__(self, name):
+ try:
+ value = self._items[name]
+ except KeyError:
+ return getattr(Reference(self), name)
+ return value
+
+ def __setattr__(self, name, what):
+ self._items[name] = what
+
+ def __getitem__(self, name):
+ return self._items[name]
+
+ def __setitem__(self, key, what):
+ if "__" in key:
+ base, rest = key.split("__", 1)
+ if base in self._items:
+ self._items[base][rest] = what
+
+ self._items[key] = what
+
+ def __iter__(self):
+ return iter(self._items)
+
+ def __call__(self, **kw):
+ new = self.__class__(self.table, self._items)
+ for k, v in kw.items():
+ new[k] = v
+ return new
+
+ @classmethod
+ def as_dict(cls, instance):
+ return instance._items.copy()
+
+ @classmethod
+ def unresolved(cls, instance):
+ """
+ Return all attributes referring to other Insertables, which must be
+ resolved before this insertable can be inserted.
+ """
+ return ((k, v) for k, v in instance._items.items() if isinstance(v, Reference))
+
+ def is_resolvable(self, resolved: set["Insertable"]) -> bool:
+ for item in self._items.values():
+ if isinstance(item, Insertable) and item not in resolved:
+ print(f"Can't resolve {self}: depends on {item}")
+ return False
+
+ elif isinstance(item, Reference) and item._source not in resolved:
+ print(f"Can't resolve {self}: depends on ref {item._source}")
+ return False
+
+ return True
+
+ @classmethod
+ def map_type(cls, type_, table):
+ """
+ Map a type to a specific table
+ """
+ cls._lookup[type_] = table
+
+ @classmethod
+ def self(cls):
+ """
+ Return an unresolved object that points to an attribute on the
+ containing object. Used when you have a parent-child relationship, for
+ example::
+
+ post = Insertable(
+ "posts",
+ title="Post title",
+ owner=Insertable("user", name="Allison Author"),
+ owner_id=Insertable.self().owner.id
+ )
+
+ """
+ return Reference.self()
+
+ @classmethod
+ def flatten(cls, insertable, minlevel=1, _level=0):
+ for k in insertable:
+ v = insertable[k]
+ if isinstance(v, Insertable):
+ if _level + 1 >= minlevel:
+ yield from cls.flatten(v, minlevel, _level + 1)
+ if _level >= minlevel:
+ yield insertable
+
+ @classmethod
+ def sort_insertables(
+ cls, items: dict[str, "Insertable"]
+ ) -> list[tuple[str | None, "Insertable"]]:
+
+ unresolved: set[tuple[str|None, Insertable]] = (
+ set(items.items())
+ | {
+ (None, i)
+ for i in chain.from_iterable(
+ Insertable.flatten(i, minlevel=1) for i in items.values()
+ )
+ }
+ )
+
+ ordering: list[tuple[str | None, Insertable]] = []
+ resolved: set[Insertable] = set()
+
+ while unresolved:
+ for k, ob in unresolved:
+ if isinstance(ob, Insertable) and ob.is_resolvable(resolved):
+ resolved.add(ob)
+ ordering.append((k, ob))
+ old_unresolved = unresolved
+ unresolved = unresolved - set(ordering)
+ if old_unresolved == unresolved:
+ raise ValueError("Cannot resolve dependencies")
+
+ return ordering
+
+ @classmethod
+ @contextlib.contextmanager
+ def _insert_dict(cls, conn, items: dict[str, "Insertable"]):
+ out = {}
+ resolved: dict[Insertable, dict[str, Any]] = {}
+ ordered = cls.sort_insertables(items)
+
+ fixture = SQLFixture(conn)
+ with contextlib.ExitStack() as stack:
+ for name, insertable in ordered:
+ insert_values = {}
+ relations = {}
+ for k in insertable:
+ v = insertable[k]
+ if isinstance(v, Insertable):
+ relations[k] = resolved[v]
+ elif isinstance(v, Reference):
+ insert_values[k] = v.resolve(resolved[v.source(insertable)])
+ else:
+ insert_values[k] = v
+
+ if insertable in resolved:
+ row = resolved[insertable]
+ else:
+ row = stack.enter_context(
+ fixture.insert(insertable.table, insert_values)
+ )
+ resolved[insertable] = row
+ for k, v in relations.items():
+ setattr(row, k, v)
+ if name:
+ out[name] = row
+
+ yield out
+
+ @classmethod
+ @contextlib.contextmanager
+ def _insert_class(cls, conn, items: type):
+ instance = items()
+ data = {k: v for k, v in inspect.getmembers(items) if isinstance(v, Insertable)}
+ with cls._insert_dict(conn, data) as inserted:
+ for k in inserted:
+ setattr(instance, k, inserted[k])
+ yield instance
+
+ @classmethod
+ @contextlib.contextmanager
+ def _insert_sequence(cls, conn, items: Iterable["Insertable"]):
+ data = {str(ix): ins for ix, ins in enumerate(items)}
+ with cls._insert_dict(conn, data) as inserted:
+ inserted_with_keys = ((int(key), value) for key, value in inserted.items())
+ yield [value for _, value in sorted(inserted_with_keys)]
+
+ @classmethod
+ @contextlib.contextmanager
+ def insert(
+ cls, conn, items: Iterable["Insertable"] | dict[str, "Insertable"] | type
+ ):
+ """
+ Create a SQLFixture that inserts the given Insertables,
+ and returns the inserted rows as named_tuples.
+ """
+
+ if isinstance(items, type):
+ with cls._insert_class(conn, items) as result:
+ yield result
+
+ elif isinstance(items, dict):
+ with cls._insert_dict(conn, items) as result:
+ yield result
+
+ else:
+ with cls._insert_sequence(conn, items) as result:
+ yield result
+
+
+class Reference:
+ """
+ A reference to an unresolved object or property.
+ """
+
+ #: Indicate that the source is the parent object itself
+ SELF = object()
+
+ def __init__(self, source):
+ self._source = source
+ self._attrs = []
+
+ def __repr__(self):
+ source = "self" if self._source is self.SELF else self._source
+ return (
+ f"<{self.__class__.__name__} "
+ f"{source}.{'.'.join(k for _, k in self._attrs)}>"
+ )
+
+ def __getattr__(self, key):
+ r = self.__class__(self._source)
+ r._attrs.extend(self._attrs)
+ r._attrs.append((operator.attrgetter, key))
+ return r
+
+ def __getitem__(self, key):
+ r = self.__class__(self._source)
+ r._attrs.extend(self.attrs)
+ r._attrs.append((operator.itemgetter, key))
+ return r
+
+ def source(self, ob):
+ if self._source is self.SELF:
+ return ob
+ else:
+ return self._source
+
+ def resolve(self, ob):
+ for op, key in self._attrs:
+ ob = op(key)(ob)
+ return ob
+
+ def unwrap(self, ob, depth=1) -> "Reference":
+ unresolved = self.__class__(ob)
+ for op, key in self._attrs[:depth]:
+ ob = op(key)(ob)
+ if isinstance(ob, Reference):
+ ob = ob._source
+ unresolved._source = ob
+ unresolved._attrs = self._attrs[depth:]
+ return unresolved
+
+ @classmethod
+ def self(cls):
+ return cls(cls.SELF)
A => sqlfixtures/sqlfixture.py +229 -0
@@ 0,0 1,229 @@
+import contextlib
+from uuid import uuid1
+
+from typing import Any
+from typing import Mapping
+from typing import Iterable
+from typing import Sequence
+
+from embrace.query import Query
+from embrace.query import mapobject
+
+RowType = Mapping | Sequence[tuple[str, Any]]
+
+
+class SQLFixture:
+ """
+ Usage::
+
+ conn = sqlite3.connect(...)
+ sqlf = SQLFixture(conn)
+
+ with sqlf.insert("user", {"name": "Angus", email="angus@example.com"}) as user:
+ assert user.name == "Angus"
+
+ """
+
+ class Literal(str):
+ """
+ A string to be inserted into the sql statement as a literal (no quoting)
+ """
+
+ def __init__(self, conn):
+ self.conn = conn
+
+ def _supports_returning(self):
+ savepoint_name = "_" + str(uuid1()).replace("-", "_")
+ cursor = self.conn.cursor()
+ cursor.execute("SAVEPOINT " + savepoint_name)
+ try:
+ cursor.execute(f"CREATE TEMPORARY TABLE _tmp_{savepoint_name} (x char(1))")
+ try:
+ cursor.execute(
+ f"INSERT INTO _tmp_{savepoint_name} VALUES ('a') RETURNING *"
+ )
+ return tuple(cursor.fetchone()) == ("a",)
+ except Exception:
+ return False
+ finally:
+ cursor.execute(f"ROLLBACK TO {savepoint_name}")
+
+ @contextlib.contextmanager
+ def insert(self, table: str, row: RowType|None = None, commit=False, **kwargs):
+ if row is None:
+ row = kwargs
+ else:
+ row = dict(row, **kwargs)
+ with self.insert_many(table, [row], commit=commit) as result:
+ yield result[0]
+
+ @contextlib.contextmanager
+ def insert_many(self, table: str, rows: Iterable[Mapping], commit=False):
+ def sigil(v):
+ if isinstance(v, self.Literal):
+ return ":raw:"
+ return ":"
+
+ rows = list(rows)
+ try:
+ firstrow = rows[0]
+ except IndexError:
+ return
+ columns = list(firstrow.keys())
+
+ sql = f"""
+ INSERT INTO {table} ({','.join(columns)})
+ VALUES ({','.join(f'{sigil(firstrow[col])}{col}' for col in columns)})
+ """
+ use_returning = self._supports_returning()
+ if use_returning:
+ sql = sql + " RETURNING *"
+
+ matching_rows_sql = f"""
+ SELECT * FROM {table} WHERE
+ {' OR '.join(
+ f"({' AND '.join(f'{c}={sigil(row[c])}_{ix}_{c}' for c in row)})"
+ for ix, row in enumerate(rows)
+ )}
+ """
+ matching_row_count_sql = f"SELECT count(1) FROM ({matching_rows_sql}) as _"
+
+ def get_matching_row_count():
+ return Query(matching_row_count_sql).scalar(
+ self.conn,
+ **{
+ f"_{ix}_{k}": v
+ for ix, row in enumerate(rows)
+ for k, v in row.items()
+ },
+ )
+
+ def get_matching_rows():
+ return (
+ Query(matching_rows_sql)
+ .returning(mapobject.dataclass())
+ .many(
+ self.conn,
+ **{
+ f"_{ix}_{k}": v
+ for ix, row in enumerate(rows)
+ for k, v in row.items()
+ },
+ )
+ )
+
+ if use_returning:
+ saved_row_count = None
+ insert = Query(sql).returning(mapobject.dataclass())
+ result = [insert.one(self.conn, **row) for row in rows]
+ else:
+ saved_row_count = get_matching_row_count()
+ if saved_row_count > 0:
+ raise Exception(
+ f"Table already contains rows matching {rows}. Inserting more "
+ f"would make it impossible to remove the rows on exit"
+ )
+ insert = Query(sql)
+ for row in rows:
+ insert.execute(self.conn, **row)
+ result = list(get_matching_rows())
+
+ if commit:
+ self.conn.commit()
+ try:
+ yield result
+ finally:
+ if result:
+ if commit:
+ returned_columns = result[0]._fields
+ delete = Query(
+ f"""
+ DELETE FROM {table} WHERE
+ {' AND '.join(f'{c} = :{c}' for c in returned_columns)}
+ """
+ )
+ for row in result:
+ delete.execute(self.conn, **row)
+ self.conn.commit()
+ else:
+ self.conn.rollback()
+ if (
+ not use_returning
+ and saved_row_count != get_matching_row_count()
+ ):
+ raise AssertionError(
+ f"Fixture values committed by test code: "
+ f"expected {saved_row_count} rows but got "
+ f"{get_matching_row_count()}."
+ )
+
+ @contextlib.contextmanager
+ def update(self, table, values, where, commit=False):
+ update_query = Query(
+ f"""
+ UPDATE {table} SET {', '.join(f'{name}=:value_{name}' for name in values)}
+ WHERE {' AND '.join(f'{name}=:where_{name}' for name in where)}
+ """
+ )
+ save_query = Query(
+ f"""
+ SELECT {', '.join(values)}
+ FROM {table}
+ WHERE {' AND '.join(f'{name}=:where_{name}' for name in where)}
+ """
+ )
+
+ def merge(*ds):
+ result = {}
+ for d in ds:
+ result.update(d)
+ return result
+
+ def prefix_keys(prefix, d):
+ return {f"{prefix}{k}": v for k, v in d.items()}
+
+ def get_save_values():
+ return dict(
+ zip(
+ values.keys(),
+ save_query.one(self.conn, **prefix_keys("where_", where)),
+ )
+ )
+
+ saved = get_save_values()
+ update_query.execute(
+ self.conn,
+ **merge(
+ prefix_keys(
+ "value_",
+ values,
+ ),
+ prefix_keys("where_", where),
+ ),
+ )
+ try:
+ if commit:
+ self.conn.commit()
+ yield
+ finally:
+ if commit:
+ update_query.execute(
+ self.conn,
+ **merge(
+ prefix_keys("value_", saved),
+ prefix_keys(
+ "where_",
+ merge(
+ where, {k: v for k, v in values.items() if k in where}
+ ),
+ ),
+ ),
+ )
+ self.conn.commit()
+ else:
+ self.conn.rollback()
+ if saved != get_save_values():
+ raise AssertionError(
+ f"Fixture values committed by test code: "
+ f"expected {saved!r} but got {get_save_values()!r}"
+ )
A => tox.ini +13 -0
@@ 0,0 1,13 @@
+[tox]
+env_list =
+ py310
+ py311
+ type
+
+[testenv]
+deps = pytest
+commands = pytest tests
+
+[testenv:type]
+deps = mypy
+commands = mypy src