evilchili/grung-db
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Adding query to unique constraint error

Browse Source
This commit is contained in:
evilchili 2025-10-03 16:43:24 -07:00
parent 1fe0d7bbdf
commit ef16f0a518
2 changed files with 13 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
src/grung/db.py
View File

@ -51,22 +51,18 @@ class RecordTable(table.Table):
self._check_unique(document) self._check_unique(document)
def _check_unique(self, document) -> bool: def _check_unique(self, document) -> bool:
matches = [ matches = []
dict(match) queries = reduce(
for match in super().search( ior,
reduce( [
ior, Query()[field.name].matches(document[field.name], flags=re.IGNORECASE)
[ for field in document._metadata.fields.values()
Query()[field.name].matches(document[field.name], flags=re.IGNORECASE) if field.unique
for field in document._metadata.fields.values() ],
if field.unique )
], matches = [dict(match) for match in super().search(queries) if match.doc_id != document.doc_id]
)
)
if match.doc_id != document.doc_id
]
if matches != []: if matches != []:
raise UniqueConstraintError(document, matches) raise UniqueConstraintError(document, queries, matches)
class GrungDB(TinyDB): class GrungDB(TinyDB):

3
src/grung/exceptions.py
View File

@ -3,10 +3,11 @@ class UniqueConstraintError(Exception):
Thrown when a db write operation cannot complete due to a field's unique constraint. Thrown when a db write operation cannot complete due to a field's unique constraint.
""" """
def __init__(self, document, collisions): def __init__(self, document, query, collisions):
super().__init__( super().__init__(
"\n" "\n"
f" * Record: {dict(document)}\n" f" * Record: {dict(document)}\n"
f" * Query: {query}\n"
f" * Error: Unique constraint failure\n" f" * Error: Unique constraint failure\n"
" * The record matches the following existing records:\n\n" + "\n".join(str(c) for c in collisions) " * The record matches the following existing records:\n\n" + "\n".join(str(c) for c in collisions)
) )