mirror of
https://github.com/simonw/sqlite-utils.git
synced 2026-11-08 03:08:41 +01:00
Merge 65b3fc8789 into 9c66bafa52
This commit is contained in:
commit
581fd944b2
3 changed files with 106 additions and 1 deletions
|
|
@ -652,6 +652,13 @@ If the table has a compound primary key you can pass in the primary key values a
|
||||||
|
|
||||||
>>> db.table("compound_dogs").get(("mixed", 3))
|
>>> db.table("compound_dogs").get(("mixed", 3))
|
||||||
|
|
||||||
|
You can also look a record up by one or more column values, passed as keyword arguments. This returns the first matching row, which is useful for fetching a record by a column with a unique constraint rather than by its primary key::
|
||||||
|
|
||||||
|
>>> db.table("dogs").get(name="Cleo")
|
||||||
|
{'id': 1, 'age': 4, 'name': 'Cleo'}
|
||||||
|
|
||||||
|
Passing more than one keyword argument matches rows against all of them. You cannot combine a primary key value with keyword arguments in the same call.
|
||||||
|
|
||||||
If the record does not exist a ``NotFoundError`` will be raised:
|
If the record does not exist a ``NotFoundError`` will be raised:
|
||||||
|
|
||||||
.. code-block:: python
|
.. code-block:: python
|
||||||
|
|
|
||||||
|
|
@ -2387,14 +2387,44 @@ class Table(Queryable):
|
||||||
"Table-level CHECK constraints on this table."
|
"Table-level CHECK constraints on this table."
|
||||||
return [check for check in self.checks if not check.column]
|
return [check for check in self.checks if not check.column]
|
||||||
|
|
||||||
def get(self, pk_values: list | tuple | str | int) -> dict:
|
def get(self, pk_values: list | tuple | str | int = DEFAULT, **kwargs) -> dict:
|
||||||
"""
|
"""
|
||||||
Return row (as dictionary) for the specified primary key.
|
Return row (as dictionary) for the specified primary key.
|
||||||
|
|
||||||
|
Alternatively, pass one or more ``column=value`` keyword arguments to
|
||||||
|
return the first row that matches those columns. This is convenient for
|
||||||
|
looking a row up by a column with a unique constraint rather than by its
|
||||||
|
primary key::
|
||||||
|
|
||||||
|
row = table.get(name="Cleo")
|
||||||
|
|
||||||
Raises ``sqlite_utils.db.NotFoundError`` if a matching row cannot be found.
|
Raises ``sqlite_utils.db.NotFoundError`` if a matching row cannot be found.
|
||||||
|
|
||||||
:param pk_values: A single value, or a tuple of values for tables that have a compound primary key
|
:param pk_values: A single value, or a tuple of values for tables that have a compound primary key
|
||||||
|
:param kwargs: Alternatively, ``column=value`` pairs to look the row up by
|
||||||
"""
|
"""
|
||||||
|
if kwargs:
|
||||||
|
if pk_values is not DEFAULT:
|
||||||
|
raise ValueError(
|
||||||
|
"get() accepts either pk_values or column keyword "
|
||||||
|
"arguments, not both"
|
||||||
|
)
|
||||||
|
wheres = [f"{quote_identifier(column)} = ?" for column in kwargs]
|
||||||
|
rows = self.rows_where(" and ".join(wheres), list(kwargs.values()), limit=1)
|
||||||
|
try:
|
||||||
|
row = next(iter(rows))
|
||||||
|
except StopIteration:
|
||||||
|
raise NotFoundError
|
||||||
|
pks = self.pks
|
||||||
|
if all(pk in row for pk in pks):
|
||||||
|
self.last_pk = (
|
||||||
|
row[pks[0]] if len(pks) == 1 else tuple(row[pk] for pk in pks)
|
||||||
|
)
|
||||||
|
return row
|
||||||
|
if pk_values is DEFAULT:
|
||||||
|
raise TypeError(
|
||||||
|
"get() requires pk_values or one or more column keyword arguments"
|
||||||
|
)
|
||||||
if not isinstance(pk_values, (list, tuple)):
|
if not isinstance(pk_values, (list, tuple)):
|
||||||
pk_values = [pk_values]
|
pk_values = [pk_values]
|
||||||
pks = self.pks
|
pks = self.pks
|
||||||
|
|
|
||||||
|
|
@ -30,3 +30,71 @@ def test_get_not_found(argument, expected_msg, fresh_db):
|
||||||
fresh_db.table("dogs").get(argument)
|
fresh_db.table("dogs").get(argument)
|
||||||
if expected_msg is not None:
|
if expected_msg is not None:
|
||||||
assert expected_msg == excinfo.value.args[0]
|
assert expected_msg == excinfo.value.args[0]
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_by_column(fresh_db):
|
||||||
|
dogs = fresh_db.table("dogs")
|
||||||
|
dogs.insert_all(
|
||||||
|
[
|
||||||
|
{"id": 1, "name": "Cleo", "age": 4},
|
||||||
|
{"id": 2, "name": "Pancakes", "age": 2},
|
||||||
|
],
|
||||||
|
pk="id",
|
||||||
|
)
|
||||||
|
assert dogs.get(name="Pancakes") == {"id": 2, "name": "Pancakes", "age": 2}
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_by_multiple_columns(fresh_db):
|
||||||
|
dogs = fresh_db.table("dogs")
|
||||||
|
dogs.insert_all(
|
||||||
|
[
|
||||||
|
{"id": 1, "name": "Cleo", "age": 4},
|
||||||
|
{"id": 2, "name": "Cleo", "age": 2},
|
||||||
|
],
|
||||||
|
pk="id",
|
||||||
|
)
|
||||||
|
assert dogs.get(name="Cleo", age=2)["id"] == 2
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_by_column_sets_last_pk(fresh_db):
|
||||||
|
dogs = fresh_db.table("dogs")
|
||||||
|
dogs.insert({"id": 5, "name": "Cleo", "age": 4}, pk="id")
|
||||||
|
dogs.get(name="Cleo")
|
||||||
|
assert dogs.last_pk == 5
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_by_column_compound_pk_sets_last_pk(fresh_db):
|
||||||
|
table = fresh_db.table("records")
|
||||||
|
table.insert_all(
|
||||||
|
[{"a": 1, "b": 2, "note": "x"}, {"a": 1, "b": 3, "note": "y"}],
|
||||||
|
pk=("a", "b"),
|
||||||
|
)
|
||||||
|
assert table.get(note="y")["b"] == 3
|
||||||
|
assert table.last_pk == (1, 3)
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_by_column_on_rowid_table(fresh_db):
|
||||||
|
dogs = fresh_db.table("dogs")
|
||||||
|
dogs.insert({"name": "Cleo", "age": 4})
|
||||||
|
assert dogs.get(name="Cleo")["name"] == "Cleo"
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_by_column_not_found(fresh_db):
|
||||||
|
dogs = fresh_db.table("dogs")
|
||||||
|
dogs.insert({"id": 1, "name": "Cleo", "age": 4}, pk="id")
|
||||||
|
with pytest.raises(NotFoundError):
|
||||||
|
dogs.get(name="Pancakes")
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_rejects_pk_and_column(fresh_db):
|
||||||
|
dogs = fresh_db.table("dogs")
|
||||||
|
dogs.insert({"id": 1, "name": "Cleo", "age": 4}, pk="id")
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
dogs.get(1, name="Cleo")
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_requires_argument(fresh_db):
|
||||||
|
dogs = fresh_db.table("dogs")
|
||||||
|
dogs.insert({"id": 1, "name": "Cleo", "age": 4}, pk="id")
|
||||||
|
with pytest.raises(TypeError):
|
||||||
|
dogs.get()
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue