remote/sqlite-utils
1
0
Fork 0
mirror of https://github.com/simonw/sqlite-utils.git synced 2026-09-17 05:54:23 +02:00
Code Issues Projects Releases Packages Wiki Activity

black formatting

Browse source
This commit is contained in:
David Kane 2020-11-16 00:32:27 +00:00
commit 3dba2029b5
4 changed files with 84 additions and 62 deletions
Download patch file Download diff file Expand all files Collapse all files

73
sqlite_utils/db.py
View file

@ -67,31 +67,37 @@ class ForeignKey:
self.other_column = (other_column,)
elif isinstance(other_column, (tuple, list)):
self.other_column = tuple(other_column)
@property
def column_str(self):
return ",".join(["[{}]".format(c) for c in self.column])
@property
def other_column_str(self):
return ",".join(["[{}]".format(c) for c in self.other_column])
def __eq__(self, other):
if isinstance(other, ForeignKey):
return all((
self.table == other.table,
self.column == other.column,
self.other_table == other.other_table,
self.other_column == other.other_column,
))
return all(
(
self.table == other.table,
self.column == other.column,
self.other_table == other.other_table,
self.other_column == other.other_column,
)
)
return False
def __lt__(self, other):
if isinstance(other, ForeignKey):
return ((self.table, self.column, self.other_table, self.other_column) <
(other.table, other.column, other.other_table, other.other_column))
return (self.table, self.column, self.other_table, self.other_column) < (
other.table,
other.column,
other.other_table,
other.other_column,
)
return False
def __repr__(self):
return "ForeignKey({table}({column}), {other_table}({other_column}))".format(
table=self.table,
@ -102,15 +108,16 @@ class ForeignKey:
@property
def sql(self):
return "FOREIGN KEY({column}) REFERENCES [{other_table}]({other_column})".format(
table=self.table,
column=self.column_str,
other_table=self.other_table,
other_column=self.other_column_str,
return (
"FOREIGN KEY({column}) REFERENCES [{other_table}]({other_column})".format(
table=self.table,
column=self.column_str,
other_table=self.other_table,
other_column=self.other_column_str,
)
)
DEFAULT = object()
COLUMN_TYPE_MAPPING = {
@ -416,12 +423,8 @@ class Database:
# Soundness check foreign_keys point to existing tables
for fk in foreign_keys:
for oc in fk.other_column:
if not any(
c for c in self[fk.other_table].columns if c.name == oc
):
raise AlterError(
"No such column: {}.{}".format(fk.other_table, oc)
)
if not any(c for c in self[fk.other_table].columns if c.name == oc):
raise AlterError("No such column: {}.{}".format(fk.other_table, oc))
column_defs = []
# ensure pk is a tuple
@ -558,10 +561,7 @@ class Database:
if not self[fk.other_table].exists():
raise AlterError("No such other_table: {}".format(fk.other_table))
for c in fk.other_column:
if (
c != "rowid"
and c not in self[fk.other_table].columns_dict
):
if c != "rowid" and c not in self[fk.other_table].columns_dict:
raise AlterError(
"No such other_column: {} in {}".format(c, fk.other_table)
)
@ -613,9 +613,7 @@ class Database:
def index_foreign_keys(self):
for table_name in self.table_names():
table = self[table_name]
existing_indexes = {
tuple(i.columns) for i in table.indexes
}
existing_indexes = {tuple(i.columns) for i in table.indexes}
for fk in table.foreign_keys:
if fk.column not in existing_indexes:
table.create_index(fk.column)
@ -769,12 +767,15 @@ class Table(Queryable):
fks[id]["column"].append(from_)
fks[id]["other_table"] = table_name
fks[id]["other_column"].append(to_)
return [ForeignKey(
table=fk['table'],
column=tuple(fk['column']),
other_table=fk['other_table'],
other_column=tuple(fk['other_column']),
) for fk in fks.values()]
return [
ForeignKey(
table=fk["table"],
column=tuple(fk["column"]),
other_table=fk["other_table"],
other_column=tuple(fk["other_column"]),
)
for fk in fks.values()
]
@property
def virtual_table_using(self):

9
tests/test_create.py
View file

@ -355,7 +355,10 @@ def test_add_foreign_key(fresh_db):
assert isinstance(t, Table) and t.name == "books"
assert [
ForeignKey(
table="books", column=("author_id",), other_table="authors", other_column=("id",)
table="books",
column=("author_id",),
other_table="authors",
other_column=("id",),
)
] == fresh_db["books"].foreign_keys
@ -364,9 +367,9 @@ def test_add_compound_foreign_key(fresh_db):
fresh_db["authors"].insert_all(
[
{"id": 1, "person_id": 1, "name": "Sally"},
{"id": 2, "person_id": 2, "name": "Asheesh"}
{"id": 2, "person_id": 2, "name": "Asheesh"},
],
pk=("id", "person_id")
pk=("id", "person_id"),
)
fresh_db["books"].insert_all(
[

4
tests/test_extract.py
View file

@ -28,7 +28,9 @@ def test_extract_single_column(fresh_db, table, fk_column):
" [name] TEXT,\n"
" [{}] INTEGER,\n".format(expected_fk)
+ " [end] INTEGER,\n"
+ " FOREIGN KEY([{}]) REFERENCES [{}]([id])\n".format(expected_fk, expected_table)
+ " FOREIGN KEY([{}]) REFERENCES [{}]([id])\n".format(
expected_fk, expected_table
)
+ ")"
)
assert fresh_db[expected_table].schema == (

60
tests/test_m2m.py
View file

@ -30,17 +30,25 @@ def test_insert_m2m_list(fresh_db):
assert [{"id": 1, "name": "Natalie D"}, {"id": 2, "name": "Simon W"}] == list(
humans.rows
)
assert sorted([
ForeignKey(
table="dogs_humans", column="dogs_id", other_table="dogs", other_column="id"
),
ForeignKey(
table="dogs_humans",
column="humans_id",
other_table="humans",
other_column="id",
),
]) == sorted(dogs_humans.foreign_keys)
assert (
sorted(
[
ForeignKey(
table="dogs_humans",
column="dogs_id",
other_table="dogs",
other_column="id",
),
ForeignKey(
table="dogs_humans",
column="humans_id",
other_table="humans",
other_column="id",
),
]
)
== sorted(dogs_humans.foreign_keys)
)
def test_insert_m2m_iterable(fresh_db):
@ -103,17 +111,25 @@ def test_m2m_lookup(fresh_db):
tags = fresh_db["tags"]
assert people_tags.exists()
assert tags.exists()
assert sorted([
ForeignKey(
table="people_tags",
column="people_id",
other_table="people",
other_column="id",
),
ForeignKey(
table="people_tags", column="tags_id", other_table="tags", other_column="id"
),
]) == sorted(people_tags.foreign_keys)
assert (
sorted(
[
ForeignKey(
table="people_tags",
column="people_id",
other_table="people",
other_column="id",
),
ForeignKey(
table="people_tags",
column="tags_id",
other_table="tags",
other_column="id",
),
]
)
== sorted(people_tags.foreign_keys)
)
assert [{"people_id": 1, "tags_id": 1}] == list(people_tags.rows)
assert [{"id": 1, "name": "Wahyu"}] == list(people.rows)
assert [{"id": 1, "tag": "Coworker"}] == list(tags.rows)