Match column names case-insensitively, closes #760

Column names passed to Python API methods are now resolved against the
table schema case-insensitively, mirroring how SQLite itself compares
identifiers (ASCII-only case folding). Fixes KeyError populating
last_pk from insert()/upsert(), silently ignored transform()
rename/drop/types options, duplicate column errors from
create_table(transform=True), redundant lookup() indexes, and
case-sensitive foreign key validation.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Simon Willison 2026-07-05 22:11:22 -07:00
commit 77d241959c
3 changed files with 442 additions and 42 deletions

View file

@ -29,6 +29,19 @@ Other foreign key improvements:
- Foreign keys declared as ``REFERENCES other_table`` with no explicit column are now resolved to the other table's primary key by ``table.foreign_keys``, instead of reporting ``other_column=None``.
- Fixed a ``TypeError`` when sorting ``ForeignKey`` objects where some were compound.
Case-insensitive column matching:
Column names passed to Python API methods are now matched against the table schema case-insensitively, mirroring how SQLite itself treats identifiers. Previously many methods accepted mixed-case identifiers in the SQL they generated but then failed - or silently did nothing - when performing Python-side comparisons against the schema. (:issue:`760`) Fixes include:
- ``table.insert()`` and ``table.upsert()`` now populate ``table.last_pk`` correctly when the ``pk=`` argument uses different casing to the table schema or the record keys - previously this raised a ``KeyError`` after the row had already been written.
- Upserts no longer raise or misbehave when the casing of ``pk=`` differs from the casing of the record keys. The primary key columns are correctly excluded from the generated ``DO UPDATE SET`` clause.
- ``table.transform()`` arguments ``types=``, ``rename=``, ``drop=``, ``pk=``, ``not_null=``, ``defaults=``, ``column_order=`` and ``drop_foreign_keys=`` all resolve column names case-insensitively. Previously options like ``rename={"name": "title"}`` against a column called ``Name`` were silently ignored.
- ``db.create_table(..., transform=True)`` now recognizes existing columns that differ only by case, instead of attempting to add them again and failing with ``duplicate column name``. The casing used in the existing schema is preserved.
- ``table.lookup()`` returns the primary key value even if ``pk=`` casing differs from the schema, and recognizes existing unique indexes case-insensitively instead of creating redundant ones.
- ``table.extract()`` and ``table.convert()`` - including ``multi=True`` and ``output=`` - accept column names in any casing.
- Foreign key columns are validated and recorded using the casing of the actual schema columns, in ``foreign_keys=`` when creating tables, ``db.add_foreign_keys()``, ``table.add_foreign_key()`` and ``table.add_column(fk_col=...)``. Duplicate foreign key detection is also case-insensitive.
- ``table.create()`` with ``pk=``, ``not_null=``, ``defaults=`` or ``column_order=`` referencing columns using different casing no longer creates an unwanted extra primary key column or raises a ``ValueError``.
.. _v4_0rc2:
4.0rc2 (2026-07-04)

View file

@ -85,6 +85,34 @@ def quote_identifier(identifier: str) -> str:
return '"{}"'.format(identifier.replace('"', '""'))
_IDENTIFIER_CASEFOLD = str.maketrans(
"ABCDEFGHIJKLMNOPQRSTUVWXYZ", "abcdefghijklmnopqrstuvwxyz"
)
def fold_identifier_case(identifier: str) -> str:
"""
Lowercase an identifier using the same rules SQLite uses - only ASCII
characters are folded, other characters are left unchanged.
"""
return identifier.translate(_IDENTIFIER_CASEFOLD)
def resolve_casing(name: str, candidates: Iterable[str]) -> str:
"""
SQLite treats identifiers as case-insensitive. Return the entry in
``candidates`` that matches ``name`` case-insensitively, preferring an
exact match. If nothing matches, return ``name`` unchanged.
"""
if name in candidates:
return name
folded = fold_identifier_case(name)
for candidate in candidates:
if fold_identifier_case(candidate) == folded:
return candidate
return name
pd: Any = None
try:
pd = importlib.import_module("pandas")
@ -1263,6 +1291,48 @@ class Database:
)
return fks
def _resolve_foreign_key_casing(
self, fk: ForeignKey, columns: Iterable[str]
) -> ForeignKey:
"""
Return ``fk`` with its column references resolved to match the casing
of the actual columns. ``columns`` provides the column names of
``fk.table``, which may be a table that is still being created.
"""
resolved_columns = tuple(resolve_casing(c, columns) for c in fk.columns)
if fk.other_table == fk.table:
other_candidates: Iterable[str] = columns
else:
other_candidates = self[fk.other_table].columns_dict
resolved_other_columns = tuple(
resolve_casing(c, other_candidates) for c in fk.other_columns
)
if (
resolved_columns == fk.columns
and resolved_other_columns == fk.other_columns
):
return fk
if fk.is_compound:
return ForeignKey(
fk.table,
None,
fk.other_table,
None,
columns=resolved_columns,
other_columns=resolved_other_columns,
is_compound=True,
on_delete=fk.on_delete,
on_update=fk.on_update,
)
return ForeignKey(
fk.table,
resolved_columns[0],
fk.other_table,
resolved_other_columns[0],
on_delete=fk.on_delete,
on_update=fk.on_update,
)
def create_table_sql(
self,
name: str,
@ -1296,11 +1366,14 @@ class Database:
"""
if hash_id_columns and (hash_id is None):
hash_id = "id"
foreign_keys = self.resolve_foreign_keys(name, foreign_keys or [])
resolved_fks: List[ForeignKey] = [
self._resolve_foreign_key_casing(fk, columns)
for fk in self.resolve_foreign_keys(name, foreign_keys or [])
]
# Compound foreign keys are rendered as table-level constraints;
# single-column ones as inline REFERENCES on their column
foreign_keys_by_column = {
fk.column: fk for fk in foreign_keys if not fk.is_compound
fk.column: fk for fk in resolved_fks if not fk.is_compound
}
# any extracts will be treated as integer columns with a foreign key
extracts = resolve_extracts(extracts)
@ -1315,8 +1388,10 @@ class Database:
name, extract_column, extract_table, "id"
)
# Soundness check not_null, and defaults if provided
not_null = not_null or set()
defaults = defaults or {}
not_null = {resolve_casing(n, columns) for n in not_null or set()}
defaults = {resolve_casing(n, columns): v for n, v in (defaults or {}).items()}
if column_order is not None:
column_order = [resolve_casing(c, columns) for c in column_order]
if not columns:
raise ValueError("Tables must have at least one column")
if not all(n in columns for n in not_null):
@ -1342,7 +1417,7 @@ class Database:
column_items.insert(0, (hash_id, str))
pk = hash_id
# Soundness check foreign_keys point to existing tables
for fk in foreign_keys:
for fk in resolved_fks:
for other_column in fk.other_columns:
if fk.other_table == name and columns.get(other_column):
continue
@ -1356,12 +1431,14 @@ class Database:
column_defs = []
# ensure pk is a tuple
single_pk = None
if isinstance(pk, list) and len(pk) == 1 and isinstance(pk[0], str):
if isinstance(pk, (list, tuple)) and len(pk) == 1 and isinstance(pk[0], str):
pk = pk[0]
if isinstance(pk, str):
single_pk = pk
single_pk = pk = resolve_casing(pk, [c[0] for c in column_items])
if pk not in [c[0] for c in column_items]:
column_items.insert(0, (pk, int))
elif pk:
pk = [resolve_casing(p, [c[0] for c in column_items]) for p in pk]
for column_name, column_type in column_items:
column_extras = []
if column_name == single_pk:
@ -1402,7 +1479,7 @@ class Database:
)
# Compound foreign keys become table-level FOREIGN KEY constraints
column_names = [c[0] for c in column_items]
for fk in foreign_keys:
for fk in resolved_fks:
if not fk.is_compound:
continue
missing = [c for c in fk.columns if c not in column_names]
@ -1483,6 +1560,11 @@ class Database:
should_transform = False
# First add missing columns and figure out columns to drop
existing_columns = table.columns_dict
# Match existing columns case-insensitively, the way SQLite does
columns = {
resolve_casing(col_name, existing_columns): col_type
for col_name, col_type in columns.items()
}
missing_columns = dict(
(col_name, col_type)
for col_name, col_type in columns.items()
@ -1506,18 +1588,28 @@ class Database:
current_pks = table.pks
desired_pk = None
if isinstance(pk, str):
desired_pk = [pk]
desired_pk = [resolve_casing(pk, existing_columns)]
elif pk:
desired_pk = list(pk)
desired_pk = [resolve_casing(p, existing_columns) for p in pk]
if desired_pk and current_pks != desired_pk:
should_transform = True
# Any not-null changes?
current_not_null = {c.name for c in table.columns if c.notnull}
desired_not_null = set(not_null) if not_null else set()
desired_not_null = (
{resolve_casing(n, existing_columns) for n in not_null}
if not_null
else set()
)
if current_not_null != desired_not_null:
should_transform = True
# How about defaults?
if defaults and defaults != table.default_values:
if (
defaults
and {
resolve_casing(c, existing_columns): v for c, v in defaults.items()
}
!= table.default_values
):
should_transform = True
# Only run .transform() if there is something to do
if should_transform:
@ -1671,12 +1763,15 @@ class Database:
is_compound=True,
)
table = fk_object.table
columns = fk_object.columns
other_table = fk_object.other_table
other_columns = fk_object.other_columns
if not self.table(table).exists():
raise AlterError("No such table: {}".format(table))
table_obj = self.table(table)
fk_object = self._resolve_foreign_key_casing(
fk_object, table_obj.columns_dict
)
columns = fk_object.columns
other_columns = fk_object.other_columns
for column in columns:
if column not in table_obj.columns_dict:
raise AlterError("No such column: {} in {}".format(column, table))
@ -1693,12 +1788,16 @@ class Database:
)
)
# We will silently skip foreign keys that exist already
columns_folded = tuple(fold_identifier_case(c) for c in columns)
other_columns_folded = tuple(fold_identifier_case(c) for c in other_columns)
if not any(
fk
for fk in table_obj.foreign_keys
if fk.columns == columns
and fk.other_table == other_table
and fk.other_columns == other_columns
if tuple(fold_identifier_case(c) for c in fk.columns) == columns_folded
and fold_identifier_case(fk.other_table)
== fold_identifier_case(other_table)
and tuple(fold_identifier_case(c) for c in fk.other_columns)
== other_columns_folded
):
foreign_keys_to_create.append(fk_object)
@ -2438,6 +2537,31 @@ class Table(Queryable):
rename = rename or {}
drop = drop or set()
# Resolve column references against the existing schema, matching
# case-insensitively the way SQLite does
existing_columns = self.columns_dict
types = {resolve_casing(c, existing_columns): t for c, t in types.items()}
rename = {resolve_casing(c, existing_columns): v for c, v in rename.items()}
drop = {resolve_casing(c, existing_columns) for c in drop}
if pk is not DEFAULT and pk is not None:
if isinstance(pk, str):
pk = resolve_casing(pk, existing_columns)
else:
pk = [resolve_casing(p, existing_columns) for p in pk]
if isinstance(not_null, dict):
not_null = {
resolve_casing(c, existing_columns): v
for c, v in cast(Dict[str, Any], not_null).items()
}
elif isinstance(not_null, set):
not_null = {resolve_casing(c, existing_columns) for c in not_null}
if defaults is not None:
defaults = {
resolve_casing(c, existing_columns): v for c, v in defaults.items()
}
if column_order is not None:
column_order = [resolve_casing(c, existing_columns) for c in column_order]
create_table_foreign_keys: List[ForeignKeyIndicator] = []
if foreign_keys is not None:
@ -2452,28 +2576,37 @@ class Table(Queryable):
create_table_foreign_keys.extend(foreign_keys)
else:
# Construct foreign_keys from current, plus add_foreign_keys, minus drop_foreign_keys
# Bind fresh names here - type checkers do not narrow captured
# variables inside nested functions, so the closures would
# otherwise see the Optional declared types of drop and rename
dropped_columns = drop
renamed_columns = rename
# The casing of columns in a foreign key definition can differ
# from the casing of the columns themselves, so these comparisons
# are all case-folded
dropped_columns_folded = {fold_identifier_case(c) for c in drop}
renamed_columns_folded = {
fold_identifier_case(k): v for k, v in rename.items()
}
def fk_should_be_dropped(fk: ForeignKey) -> bool:
fk_columns_folded = tuple(fold_identifier_case(c) for c in fk.columns)
if drop_foreign_keys is not None:
for spec in drop_foreign_keys:
if isinstance(spec, str):
# A column name matches any foreign key it participates in
if spec in fk.columns:
if fold_identifier_case(spec) in fk_columns_folded:
return True
elif tuple(spec) == fk.columns:
elif (
tuple(fold_identifier_case(s) for s in spec)
== fk_columns_folded
):
# A tuple/list must match a compound key's columns exactly
return True
# Dropping any of a foreign key's columns drops the whole key
return any(column in dropped_columns for column in fk.columns)
return any(
column in dropped_columns_folded for column in fk_columns_folded
)
def fk_with_renamed_columns(fk: ForeignKey) -> ForeignKey:
columns = tuple(
renamed_columns.get(column) or column for column in fk.columns
renamed_columns_folded.get(fold_identifier_case(column)) or column
for column in fk.columns
)
if fk.is_compound:
return ForeignKey(
@ -2662,6 +2795,8 @@ class Table(Queryable):
rename = rename or {}
if isinstance(columns, str):
columns = [columns]
columns = [resolve_casing(c, self.columns_dict) for c in columns]
rename = {resolve_casing(k, self.columns_dict): v for k, v in rename.items()}
if not set(columns).issubset(self.columns_dict.keys()):
raise InvalidColumns(
"Invalid columns {} for table with columns {}".format(
@ -2853,6 +2988,7 @@ class Table(Queryable):
raise AlterError("table '{}' does not exist".format(fk))
# if fk_col specified, must be a valid column
if fk_col is not None:
fk_col = resolve_casing(fk_col, self.db[fk].columns_dict)
if fk_col not in self.db[fk].columns_dict:
raise AlterError("table '{}' has no column {}".format(fk, fk_col))
else:
@ -2958,6 +3094,7 @@ class Table(Queryable):
:param on_update: ``ON UPDATE`` action for the foreign key.
"""
columns = (column,) if isinstance(column, str) else tuple(column)
columns = tuple(resolve_casing(c, self.columns_dict) for c in columns)
# Ensure columns exist
for col in columns:
if col not in self.columns_dict:
@ -2979,6 +3116,9 @@ class Table(Queryable):
other_columns = (other_column,)
else:
other_columns = tuple(other_column)
other_columns = tuple(
resolve_casing(c, self.db[other_table].columns_dict) for c in other_columns
)
if len(columns) != len(other_columns):
raise ValueError(
"Compound foreign key must have the same number of columns "
@ -2996,9 +3136,12 @@ class Table(Queryable):
if any(
fk
for fk in self.foreign_keys
if fk.columns == columns
and fk.other_table == other_table
and fk.other_columns == other_columns
if tuple(fold_identifier_case(c) for c in fk.columns)
== tuple(fold_identifier_case(c) for c in columns)
and fold_identifier_case(fk.other_table)
== fold_identifier_case(other_table)
and tuple(fold_identifier_case(c) for c in fk.other_columns)
== tuple(fold_identifier_case(c) for c in other_columns)
):
if ignore:
return self
@ -3547,6 +3690,7 @@ class Table(Queryable):
"""
if isinstance(columns, str):
columns = [columns]
columns = [resolve_casing(c, self.columns_dict) for c in columns]
if multi:
return self._convert_multi(
@ -3561,6 +3705,7 @@ class Table(Queryable):
if output is not None:
if len(columns) != 1:
raise ValueError("output= can only be used with a single column")
output = resolve_casing(output, self.columns_dict)
if output not in self.columns_dict:
self.add_column(output, output_type or "text")
@ -3760,6 +3905,8 @@ class Table(Queryable):
# Everything from here on is for upsert=True
pk_cols = [pk] if isinstance(pk, str) else list(pk)
# The records may use different casing for the pk columns than pk=
pk_cols = [resolve_casing(c, all_columns) for c in pk_cols]
# Every record must provide a value for every primary key column - a
# NULL primary key never matches ON CONFLICT, so the record would be
# inserted as a brand new row instead of upserted
@ -3810,10 +3957,7 @@ class Table(Queryable):
# At this point we need compatibility UPSERT for SQLite < 3.24.0
# (INSERT OR IGNORE + second UPDATE stage)
queries_and_params = []
if isinstance(pk, str):
pks = [pk]
else:
pks = pk
pks = pk_cols
self.last_pk = None
for record_values in values:
record = dict(zip(all_columns, record_values))
@ -4226,9 +4370,9 @@ class Table(Queryable):
if hash_id:
self.last_pk = row[hash_id]
elif isinstance(pk, str):
self.last_pk = row[pk]
self.last_pk = row[resolve_casing(pk, row)]
else:
self.last_pk = tuple(row[p] for p in pk)
self.last_pk = tuple(row[resolve_casing(p, row)] for p in pk)
else:
self.last_pk = self.last_rowid
else:
@ -4240,11 +4384,14 @@ class Table(Queryable):
# hash_id not supported in list mode for last_pk
pass
elif isinstance(pk, str):
pk_index = column_names.index(pk)
pk_index = column_names.index(resolve_casing(pk, column_names))
self.last_pk = first_record_list[pk_index]
else:
self.last_pk = tuple(
first_record_list[column_names.index(p)] for p in pk
first_record_list[
column_names.index(resolve_casing(p, column_names))
]
for p in pk
)
else:
first_record_dict = cast(Dict[str, Any], first_record)
@ -4252,9 +4399,12 @@ class Table(Queryable):
self.last_pk = hash_record(first_record_dict, hash_id_columns)
else:
self.last_pk = (
first_record_dict[pk]
first_record_dict[resolve_casing(pk, first_record_dict)]
if isinstance(pk, str)
else tuple(first_record_dict[p] for p in pk)
else tuple(
first_record_dict[resolve_casing(p, first_record_dict)]
for p in pk
)
)
if analyze:
@ -4399,8 +4549,12 @@ class Table(Queryable):
combined_values.update(extra_values)
if self.exists():
self.add_missing_columns([combined_values])
unique_column_sets = [set(i.columns) for i in self.indexes]
if set(lookup_values.keys()) not in unique_column_sets:
unique_column_sets = [
{fold_identifier_case(c) for c in i.columns} for i in self.indexes
]
if {
fold_identifier_case(c) for c in lookup_values
} not in unique_column_sets:
self.create_index(lookup_values.keys(), unique=True)
wheres = [
"{} = ?".format(quote_identifier(column)) for column in lookup_values
@ -4411,7 +4565,7 @@ class Table(Queryable):
)
)
try:
return rows[0][pk]
return rows[0][resolve_casing(pk, rows[0])]
except IndexError:
return self.insert(
combined_values,

233
tests/test_column_casing.py Normal file
View file

@ -0,0 +1,233 @@
"""
SQLite treats column names as case-insensitive. These tests exercise the
places where sqlite-utils performs Python-side lookups of column names
provided by the caller, which should match the schema case-insensitively.
https://github.com/simonw/sqlite-utils/issues/760
"""
import pytest
from sqlite_utils import Database
from sqlite_utils.db import ForeignKey
def test_insert_populates_last_pk_case_insensitively(fresh_db):
books = fresh_db["books"]
books.create({"Id": int, "Title": str}, pk="Id")
books.insert({"Id": 1, "Title": "One"}, pk="id")
assert books.last_pk == 1
def test_insert_populates_last_pk_compound_pk_case_insensitively(fresh_db):
books = fresh_db["books"]
books.create({"Author": str, "Position": int, "Title": str})
books.insert(
{"Author": "Sue", "Position": 1, "Title": "One"}, pk=("author", "position")
)
assert books.last_pk == ("Sue", 1)
@pytest.mark.parametrize("use_old_upsert", (False, True))
def test_upsert_pk_case_differs_from_schema(use_old_upsert):
db = Database(memory=True, use_old_upsert=use_old_upsert)
books = db["books"]
books.create({"Id": int, "Title": str}, pk="Id")
books.insert({"Id": 1, "Title": "One"})
books.upsert({"id": 1, "title": "Won"}, pk="id")
assert list(books.rows) == [{"Id": 1, "Title": "Won"}]
assert books.last_pk == 1
@pytest.mark.parametrize("use_old_upsert", (False, True))
def test_upsert_record_key_case_differs_from_pk(use_old_upsert):
# all_columns comes from the record keys, pk= from the caller
db = Database(memory=True, use_old_upsert=use_old_upsert)
books = db["books"]
books.create({"Id": int, "Title": str}, pk="Id")
books.upsert({"ID": 1, "Title": "One"}, pk="id")
assert list(books.rows) == [{"Id": 1, "Title": "One"}]
assert books.last_pk == 1
def test_upsert_inferred_pk_case_differs_from_record_keys(fresh_db):
# pk is inferred from the existing schema as "Id", records use "id"
books = fresh_db["books"]
books.create({"Id": int, "Title": str}, pk="Id")
books.upsert({"id": 1, "title": "One"})
assert list(books.rows) == [{"Id": 1, "Title": "One"}]
assert books.last_pk == 1
def test_upsert_list_mode_pk_case_insensitive(fresh_db):
books = fresh_db["books"]
books.create({"Id": int, "Title": str}, pk="Id")
books.upsert_all([["id", "title"], [1, "One"]], pk="Id")
assert list(books.rows) == [{"Id": 1, "Title": "One"}]
assert books.last_pk == 1
def test_lookup_pk_case_insensitive(fresh_db):
fresh_db["species"].create({"ID": int, "Name": str}, pk="ID")
fresh_db["species"].insert({"ID": 5, "Name": "Palm"})
fresh_db["species"].create_index(["Name"], unique=True)
assert fresh_db["species"].lookup({"Name": "Palm"}, pk="id") == 5
def test_lookup_does_not_create_redundant_index(fresh_db):
fresh_db["species"].create({"id": int, "Name": str}, pk="id")
fresh_db["species"].create_index(["Name"], unique=True)
fresh_db["species"].lookup({"name": "Palm"})
assert len(fresh_db["species"].indexes) == 1
def test_create_table_transform_same_columns_different_case(fresh_db):
fresh_db["t"].create({"Name": str, "Age": int})
fresh_db["t"].insert({"Name": "Cleo", "Age": 5})
fresh_db.create_table("t", {"name": str, "age": int}, transform=True)
# Schema casing is preserved - SQLite considers these the same columns
assert fresh_db["t"].columns_dict == {"Name": str, "Age": int}
assert list(fresh_db["t"].rows) == [{"Name": "Cleo", "Age": 5}]
def test_create_table_transform_case_insensitive_with_changes(fresh_db):
fresh_db["t"].create({"Name": str, "Age": int})
fresh_db.create_table("t", {"name": str, "age": str, "size": int}, transform=True)
# age changed type, size added, Name untouched
assert fresh_db["t"].columns_dict == {"Name": str, "Age": str, "size": int}
def test_transform_types_case_insensitive(fresh_db):
fresh_db["t"].create({"Name": str, "Age": str})
fresh_db["t"].transform(types={"age": int})
assert fresh_db["t"].columns_dict == {"Name": str, "Age": int}
def test_transform_rename_case_insensitive(fresh_db):
fresh_db["t"].create({"Name": str})
fresh_db["t"].transform(rename={"name": "title"})
assert fresh_db["t"].columns_dict == {"title": str}
def test_transform_drop_case_insensitive(fresh_db):
fresh_db["t"].create({"Name": str, "Age": int})
fresh_db["t"].transform(drop=["name"])
assert fresh_db["t"].columns_dict == {"Age": int}
def test_transform_not_null_and_defaults_case_insensitive(fresh_db):
fresh_db["t"].create({"Name": str, "Age": int})
fresh_db["t"].transform(not_null={"name"}, defaults={"age": 3})
columns = {c.name: c for c in fresh_db["t"].columns}
assert columns["Name"].notnull
assert fresh_db["t"].default_values == {"Age": 3}
def test_transform_pk_case_insensitive(fresh_db):
fresh_db["t"].create({"Id": int, "Name": str})
fresh_db["t"].transform(pk="id")
assert fresh_db["t"].pks == ["Id"]
assert fresh_db["t"].columns_dict == {"Id": int, "Name": str}
def test_transform_drop_foreign_keys_case_insensitive(fresh_db):
fresh_db["parent"].create({"Id": int}, pk="Id")
fresh_db["child"].create(
{"id": int, "Parent_ID": int},
pk="id",
foreign_keys=[("Parent_ID", "parent", "Id")],
)
fresh_db["child"].transform(drop_foreign_keys=["parent_id"])
assert fresh_db["child"].foreign_keys == []
def test_add_foreign_key_case_insensitive(fresh_db):
fresh_db["parent"].create({"Id": int}, pk="Id")
fresh_db["child"].create({"id": int, "Parent_ID": int}, pk="id")
fresh_db["child"].add_foreign_key("parent_id", "parent", "id")
fks = fresh_db["child"].foreign_keys
assert len(fks) == 1
# The foreign key should use the schema casing of the columns
assert fks[0].column == "Parent_ID"
assert fks[0].other_column == "Id"
def test_add_foreign_keys_case_insensitive(fresh_db):
fresh_db["parent"].create({"Id": int}, pk="Id")
fresh_db["child"].create({"id": int, "Parent_ID": int}, pk="id")
fresh_db.add_foreign_keys([("child", "parent_id", "parent", "id")])
fks = fresh_db["child"].foreign_keys
assert len(fks) == 1
assert fks[0].column == "Parent_ID"
assert fks[0].other_column == "Id"
def test_add_foreign_key_detects_existing_case_insensitively(fresh_db):
fresh_db["parent"].create({"Id": int}, pk="Id")
fresh_db["child"].create(
{"id": int, "Parent_ID": int},
pk="id",
foreign_keys=[("Parent_ID", "parent", "Id")],
)
# ignore=True should treat this as already existing, not add a duplicate
fresh_db["child"].add_foreign_key("parent_id", "parent", "id", ignore=True)
assert len(fresh_db["child"].foreign_keys) == 1
def test_add_column_fk_col_case_insensitive(fresh_db):
fresh_db["parent"].create({"Id": int}, pk="Id")
fresh_db["child"].create({"id": int}, pk="id")
fresh_db["child"].add_column("parent_id", int, fk="parent", fk_col="id")
fks = fresh_db["child"].foreign_keys
assert len(fks) == 1
assert fks[0].other_column == "Id"
def test_extract_case_insensitive(fresh_db):
fresh_db["trees"].insert({"id": 1, "Species": "Palm"}, pk="id")
fresh_db["trees"].extract("species")
assert fresh_db["trees"].columns_dict == {"id": int, "Species_id": int}
assert list(fresh_db["Species"].rows) == [{"id": 1, "Species": "Palm"}]
def test_convert_multi_case_insensitive(fresh_db):
fresh_db["t"].insert({"id": 1, "Name": "Cleo"}, pk="id")
fresh_db["t"].convert("name", lambda v: {"upper": v.upper()}, multi=True)
assert list(fresh_db["t"].rows) == [{"id": 1, "Name": "Cleo", "upper": "CLEO"}]
def test_convert_output_case_insensitive(fresh_db):
fresh_db["t"].insert({"id": 1, "Name": "Cleo", "Upper": None}, pk="id")
fresh_db["t"].convert("name", lambda v: v.upper(), output="upper")
assert list(fresh_db["t"].rows) == [{"id": 1, "Name": "Cleo", "Upper": "CLEO"}]
def test_create_table_sql_pk_case_insensitive(fresh_db):
fresh_db["t"].create({"Id": int, "Name": str}, pk="id")
# Should not have created an extra lowercase "id" column
assert fresh_db["t"].columns_dict == {"Id": int, "Name": str}
assert fresh_db["t"].pks == ["Id"]
def test_create_table_not_null_and_defaults_case_insensitive(fresh_db):
fresh_db["t"].create(
{"Name": str, "Age": int}, not_null={"name"}, defaults={"age": 1}
)
columns = {c.name: c for c in fresh_db["t"].columns}
assert columns["Name"].notnull
assert fresh_db["t"].default_values == {"Age": 1}
def test_create_table_foreign_keys_case_insensitive(fresh_db):
fresh_db["parent"].create({"Id": int}, pk="Id")
fresh_db["child"].create(
{"id": int, "Parent_ID": int},
pk="id",
foreign_keys=[("parent_id", "parent", "id")],
)
fks = fresh_db["child"].foreign_keys
assert fks == [
ForeignKey(
table="child", column="Parent_ID", other_table="parent", other_column="Id"
)
]