ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(import): load a site's data from spreadsheets
Some checks failed
CI / backend (push) Failing after 7s
Details
CI / naming (push) Successful in 2s
Details
CI / frontend (push) Successful in 9s
Details
CI / migrations-mysql (push) Failing after 6s
Details

Browse Source 
Adopting a site means getting its asset register in. The HTTP import API suits a
site with a source system and someone to script against it; a sister site with a
spreadsheet and no developer needs something else, and that is the common case.

FOREIGN KEYS TAKE NAMES. This is the whole design. A CSV row has to say where an
asset is, and the database stores locationid, an integer. Requiring the number
means importing locations, reading back the generated ids and pasting them into
the asset sheet - a workflow nobody finishes. Every foreign key here accepts
either a numeric id or the referenced row's name:

    assetnumber,assettypeid,statusid,locationid
    CMM-01,Measuring Tool,Active,Gage Lab

The column keeps its database name, per CONTRIBUTING.md; the value is whatever
the operator actually knows. Names resolve across files in one run, so
assets.csv can reference a location that only exists because locations.csv was
read moments earlier. A name that does not resolve is reported with its line,
column and value, not as a foreign key violation from three layers down.

Dry run is the default, and writes go into the transaction either way - the
rollback is what makes it a dry run. Skipping the writes instead made every
cross-file reference fail, which is the one thing a folder-wide check exists to
verify. Validation covers every row before anything is written, so a typo on
line 400 cannot leave 399 rows imported. Files are matched on a natural key, so
correcting a spreadsheet and re-running updates rather than duplicates.

TEMPLATES ARE GENERATED, NOT MAINTAINED. "flask csv templates" builds them from
the live schema, annotated with required/optional and which file each foreign
key refers to. The prompt for this was a hand-written template set that had
invented columns on seven of eleven tables and named a table that does not
exist, while looking entirely plausible - and described an import mechanism
(a Data Import page, a flask import-csv command) that had never existed. A test
fails the build if a generated template ever offers a column the schema lacks.

User accounts are deliberately not importable: passwords do not belong in a
spreadsheet in either direction.

Verified end to end against MySQL 5.6 - a folder dry run catching one bad
reference, the fix, the commit, and a re-run reporting updates rather than
inserts. 16 tests.
This commit is contained in:
cproudlock
2026-08-04 09:12:33 -04:00
parent f72813ed9c
commit 4a8bd138a9
7 changed files with 1075 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

142
docs/CSV-IMPORT.md Normal file
View File

@@ -0,0 +1,142 @@
# Loading a site's data from spreadsheets
For getting a new site's starting data in when you have a spreadsheet rather
than a source database to script against. No developer needed.
If the site *does* have a source system worth reading, the HTTP import API is
the better tool - see [IMPORT-ADOPTION.md](IMPORT-ADOPTION.md).
---
## The short version
```bash
cd C:\shopdb-flask # or your install directory
venv\Scripts\flask csv templates --out csv-templates
```
Fill in the templates. Then:
```bash
venv\Scripts\flask csv import --dir csv-templates
```
That **checks only** and changes nothing. It tells you what it would create and
what is wrong. When you are happy:
```bash
venv\Scripts\flask csv import --dir csv-templates --commit
```
---
## Write names, not numbers
This is the part that makes the difference. Every column that points at another
table accepts the **name** of the thing:
```
assetnumber,name,assettypeid,statusid,locationid
CMM-01,Zeiss Contura,Measuring Tool,Active,Gage Lab
MILL-07,Haas VF-2,Machine,Active,Bay 3
```
`assettypeid` gets `Measuring Tool`. `locationid` gets `Gage Lab`. You never
have to import a file, read back the numbers it generated, and paste them into
the next one.
The column keeps its database name so it matches the rest of the system, but
the value is whatever you actually know. Numeric ids still work if you have
them - useful when re-importing something this system exported.
Names resolve **across files in the same run**, so `assets.csv` can reference a
location that only exists because `locations.csv` was loaded moments earlier.
## Nothing is half-imported
Every row is checked before anything is written. If one row is wrong, nothing is
written at all - you fix the file and run it again. A mistake on line 400 never
leaves 399 rows loaded.
## Running it twice is safe
Each file is matched on a natural key - `assetnumber` for assets, `locationname`
for locations, and so on. Re-importing an edited file **updates** those rows
rather than creating second copies. Correcting a spreadsheet and re-running is
the expected workflow, not a mistake.
## What the errors look like
```
assets: 0 new, 0 updated, 1 problem(s)
line 4, column 'locationid': nothing in locations is named 'Bay 9'
- add it to locations.csv, or import that file first
```
Line, column, value, and what to do. Not a foreign key constraint violation.
---
## What you can import
Fourteen tables, in the order the importer handles them. You only need the ones
you have; skip any file you do not care about.
| Order | File | Matched on |
|---|---|---|
| 1 | `assetstatuses.csv` | `status` |
| 2 | `assettypes.csv` | `assettype` |
| 3 | `locationtypes.csv` | `locationtype` |
| 4 | `modeltypes.csv` | `modeltype` |
| 5 | `computertypes.csv` | `computertype` |
| 6 | `machinetypes.csv` | `machinetype` |
| 7 | `businessunits.csv` | `businessunit` |
| 8 | `locations.csv` | `locationname` |
| 9 | `vendors.csv` | `vendor` |
| 10 | `models.csv` | `modelnumber` |
| 11 | `operatingsystems.csv` | `osname` |
| 12 | `assets.csv` | `assetnumber` |
| 13 | `computers.csv` | `assetid` |
| 14 | `machines.csv` | `assetid` |
`--dir` handles the order for you. Use `--file` with `--table` for one file.
**User accounts are deliberately not importable.** Passwords do not belong in a
spreadsheet, in either direction. Create the first administrator through the
first-run page and the rest in the application.
## The templates are generated, not maintained
`flask csv templates` builds them from the live database schema each time. Every
column offered exists; every required one is marked; every foreign key says
which file it refers to.
This matters because the alternative does not work. A hand-written template set
was tried, and it had invented columns on seven of eleven tables and named a
table that does not exist - while looking entirely plausible. Templates that are
generated cannot drift from the schema, and a test fails the build if they ever
do.
## Editing the files
- **UTF-8**, no BOM. Excel: "CSV UTF-8 (Comma delimited)".
- Lines starting with `#` are ignored, so the notes and the example row in each
template can stay where they are.
- Booleans are `1` or `0`.
- Dates are `YYYY-MM-DD` (`2026-08-04`). `YYYY-MM-DD HH:MM:SS` also works, as do
`DD/MM/YYYY` and `MM/DD/YYYY`.
- Leave a cell **empty** for "no value". Not `NULL`, not `N/A`.
- Quote anything containing a comma: `"Bay 3, North"`.
## If it will not run
**"the 'assets' table does not exist in this database"** - the schema has not
been created. Run `flask db upgrade` first, and check `DATABASE_URL` points at
the site you meant.
**"unknown column(s): ..."** - a column that does not exist, usually from an
older template. Regenerate with `flask csv templates`; the message lists what
the table does accept.
**"required column(s) missing: ..."** - a column that must be present has been
deleted from the header. Regenerate and copy your data across.

25
docs/IMPORT-ADOPTION.md
View File

@@ -1,5 +1,30 @@
# Importing a site's legacy data
## Two routes in, and which one you want
**If the site has a spreadsheet and no developer**, use the CSV import. It is
the common case, and it needs nothing beyond the templates:
```bash
flask csv templates --out csv-templates # generated from the live schema
# fill them in
flask csv import --dir csv-templates # checks only, changes nothing
flask csv import --dir csv-templates --commit
```
Foreign keys take a NAME, not an id - write `Bay 3`, not `locationid=7`. The
importer resolves them, including across files in the same run, and a name it
cannot find is reported with the line, the column and the value. Nothing is
written unless every row passes, and re-running an edited file updates rows
rather than duplicating them. See [CSV-IMPORT.md](CSV-IMPORT.md).
**If the site has a source database to read from**, and someone able to script
against it, the HTTP import API below is the better tool: it carries the whole
history, preserves original timestamps, and handles relationships the CSV set
does not model.
---
Every adopting site has its own source database - it will not match another
site's schema. So the import is split in two layers:

13
docs/llms.txt
View File

@@ -28,6 +28,19 @@ answers "does this server carry component X" from the on-box CycloneDX SBOM.
Python is 3.14 and the wheelhouse is locked to it; an upgrade against a venv
built by a different minor version is refused by design.
## Bulk-loading a site's data
Two routes, and the right answer depends on what the site has:
- SPREADSHEET, no developer (the common case): `flask csv templates --out <dir>`
generates templates FROM THE LIVE SCHEMA, then `flask csv import --dir <dir>`
checks and `--commit` applies. Foreign keys accept the NAME of the referenced
row ('Bay 3'), not a numeric id, and resolve across files in one run. Dry run
is the default; nothing is written unless every row passes; re-importing an
edited file updates rather than duplicates. See `docs/CSV-IMPORT.md`.
- A SOURCE DATABASE to script against: the HTTP import API, `docs/IMPORT-API.md`
and `docs/IMPORT-ADOPTION.md`.
Do NOT hand-write CSV templates - generate them. User accounts are deliberately
not CSV-importable.
## Base URL
Prod (West Jefferson): `https://tsgwp00525.wjs.geaerospace.net/shopdb`
All API paths are under `/api` (e.g. `<base>/api/assets`). Dev: `http://localhost:5001`.

3
shopdb/__init__.py
View File

@@ -172,12 +172,13 @@ def register_blueprints(app: Flask):
def register_cli_commands(app: Flask):
"""Register Flask CLI commands."""
from .plugins.cli import plugin_cli
from .cli import db_cli, seed_cli, relationships_cli
from .cli import db_cli, seed_cli, relationships_cli, csv_cli
app.cli.add_command(plugin_cli)
app.cli.add_command(db_cli)
app.cli.add_command(seed_cli)
app.cli.add_command(relationships_cli)
app.cli.add_command(csv_cli)
def register_error_handlers(app: Flask):

157
shopdb/cli/__init__.py
View File

@@ -825,3 +825,160 @@ def seed_demo_clear(yes):
click.echo(click.style(
f"Removed {len(demo_ids)} demo assets and "
f"{parts_deleted} sample parts.", fg='green'))
@click.group('csv')
def csv_cli():
"""Load a site's starting data from CSV files."""
pass
@csv_cli.command('templates')
@click.option('--out', 'outdir', default='csv-templates',
help='Directory to write the templates into.')
@with_appcontext
def csv_templates(outdir):
"""Write a CSV template per importable table, generated from the schema.
Generated rather than kept by hand: a maintained template set drifts on the
next migration and does so silently, since the file still looks correct.
"""
import os
from shopdb.core.services.csvimport import IMPORTABLE, generate_template
if not os.path.isdir(outdir):
os.makedirs(outdir)
for tablename in IMPORTABLE:
path = os.path.join(outdir, tablename + '.csv')
with open(path, 'w', encoding='utf-8') as handle:
handle.write(generate_template(tablename))
click.echo(' %s' % path)
readme = os.path.join(outdir, 'README.txt')
with open(readme, 'w', encoding='utf-8') as handle:
handle.write(
'ShopDB-Flask import templates\n'
'=============================\n\n'
'Generated from the live database schema. Every column here exists;\n'
'every required column is marked.\n\n'
'Fill in the ones you need - you do not need all of them.\n\n'
'Foreign keys take a NAME or a numeric id. Write the name:\n'
' locationid -> Building 1 Bay 3\n'
' vendorid -> Haas Automation\n'
'The importer resolves it, and tells you which row and column to fix\n'
'if the name is not found.\n\n'
'Import the whole folder at once and order is handled for you:\n\n'
' flask csv import --dir . (checks only, changes nothing)\n'
' flask csv import --dir . --commit (applies)\n\n'
'Nothing is written unless every row passes, so a mistake on line 400\n'
'does not leave 399 rows half-imported.\n\n'
'User accounts are deliberately not importable here: passwords do not\n'
'belong in a spreadsheet.\n')
click.echo(' %s' % readme)
click.echo('')
click.echo(click.style('%d templates written to %s' % (len(IMPORTABLE), outdir),
fg='green', bold=True))
@csv_cli.command('import')
@click.option('--file', 'path', default=None, help='One CSV file.')
@click.option('--dir', 'directory', default=None,
help='A folder of CSVs, imported in dependency order.')
@click.option('--table', 'tablename', default=None,
help='Target table. Defaults to the file name.')
@click.option('--commit', is_flag=True, default=False,
help='Apply the changes. Without this, nothing is written.')
@with_appcontext
def csv_import(path, directory, tablename, commit):
"""Validate CSVs and, with --commit, load them.
Dry run by default. The report is the same either way, so what you review is
what you get.
"""
import os
from shopdb.extensions import db
from sqlalchemy.exc import SQLAlchemyError
from shopdb.core.services.csvimport import (
ImportError_, Resolver, dependency_order, import_csv, table_from_filename)
if not path and not directory:
raise click.UsageError('give --file or --dir')
jobs = []
if path:
jobs.append((tablename or table_from_filename(os.path.basename(path)), path))
else:
found = {}
for name in os.listdir(directory):
if not name.lower().endswith('.csv'):
continue
found[table_from_filename(name)] = os.path.join(directory, name)
ordered = dependency_order(list(found))
unknown = sorted(set(found) - set(ordered))
for name in ordered:
jobs.append((name, found[name]))
if unknown:
click.echo(click.style(
'skipping (not importable): %s' % ', '.join(unknown), fg='yellow'))
if not jobs:
click.echo('nothing to do - no CSV files found')
return
click.echo(click.style(
'Checking %d file(s)%s' % (len(jobs), '' if commit else ' - DRY RUN, nothing will be written'),
bold=True))
click.echo('')
resolver = Resolver()
results = []
failed = False
for name, filepath in jobs:
with open(filepath, 'r', encoding='utf-8-sig') as handle:
text = handle.read()
try:
# Every file is applied inside ONE transaction, so a failure part way
# through a folder rolls the whole run back rather than leaving the
# site half-populated.
result = import_csv(name, text, resolver=resolver, commit=commit)
except ImportError_ as exc:
click.echo(click.style(' %-18s %s' % (name, exc), fg='red'))
failed = True
continue
except SQLAlchemyError as exc:
# Anything the database itself refuses. The operator gets the cause
# in one line rather than a traceback they cannot act on.
db.session.rollback()
click.echo(click.style(' %-18s database error: %s'
% (name, str(exc).split(chr(10))[0]), fg='red'))
failed = True
continue
results.append(result)
colour = 'green' if result.ok else 'red'
click.echo(click.style(' ' + result.summary(), fg=colour))
for problem in result.problems[:20]:
click.echo(click.style(' %s' % problem, fg='red'))
if len(result.problems) > 20:
click.echo(click.style(' ... and %d more' % (len(result.problems) - 20), fg='red'))
if not result.ok:
failed = True
click.echo('')
if failed:
db.session.rollback()
click.echo(click.style('Nothing was imported. Fix the problems above and run again.',
fg='red', bold=True))
raise SystemExit(1)
total_new = sum(r.created for r in results)
total_upd = sum(r.updated for r in results)
if commit:
db.session.commit()
click.echo(click.style('Imported: %d new, %d updated.' % (total_new, total_upd),
fg='green', bold=True))
else:
db.session.rollback()
click.echo(click.style(
'Looks good: %d would be created, %d updated.' % (total_new, total_upd),
fg='green', bold=True))
click.echo('Run again with --commit to apply.')

500
shopdb/core/services/csvimport.py Normal file
View File

@@ -0,0 +1,500 @@
"""Bulk-load a site's starting data from CSV files.
WHY THIS EXISTS. Adopting a site means getting its existing asset register into
ShopDB. The HTTP import surface (docs/IMPORT-API.md) is the right tool when
there is a source system to read from and someone able to script against it. A
sister site with a spreadsheet and no developer needs something else, and that
is the common case.
THE DESIGN PROBLEM, which a hand-written template set gets wrong. A CSV row for
an asset has to say where the asset is. The database stores `locationid`, an
integer. Nobody types integers into a spreadsheet correctly, and requiring it
means the operator must first import locations, read back the generated ids, and
paste them into the asset sheet. That is not a workflow anyone completes.
So every foreign key here accepts EITHER a numeric id OR the referenced row's
name, and names are resolved at import time:
assetnumber,assettypeid,locationid,businessunitid
CMM-01,Measuring Tool,Building 1 Bay 3,Quality
The column keeps its database name - the convention in CONTRIBUTING.md - while
the VALUE is whatever the operator actually knows. A name that does not resolve
is a row error naming the column, the value, and where to fix it, not a foreign
key constraint violation from three layers down.
IDEMPOTENCE. Each table declares a natural key (`assetnumber`, `locationname`,
...). Re-importing a file updates the matched rows rather than duplicating them,
so a site can correct its spreadsheet and run it again - which they will.
SAFETY. Validation is a separate pass over the whole file BEFORE anything is
written, and the default is a dry run. An import either applies completely or
not at all.
"""
from __future__ import annotations
import csv
import io
import re
from collections import OrderedDict
from sqlalchemy import inspect as sqla_inspect
from shopdb.extensions import db
class ImportError_(Exception):
"""A problem with the request itself, not with a row."""
# The tables a site may load, in dependency order: a table never appears before
# something it references. `label` is the human-facing column - the one an
# operator would write in a spreadsheet, and the one used to resolve this table
# when another table points at it. `natural` is what makes a re-import an update
# rather than a duplicate.
#
# Deliberately NOT every table. Transactional and derived data (audit rows,
# reports, enforcement history) has no business arriving by spreadsheet, and
# offering it would invite someone to try.
IMPORTABLE = OrderedDict([
('assetstatuses', {'label': 'status', 'natural': 'status'}),
('assettypes', {'label': 'assettype', 'natural': 'assettype'}),
('locationtypes', {'label': 'locationtype', 'natural': 'locationtype'}),
('modeltypes', {'label': 'modeltype', 'natural': 'modeltype'}),
('computertypes', {'label': 'computertype', 'natural': 'computertype'}),
('machinetypes', {'label': 'machinetype', 'natural': 'machinetype'}),
('businessunits', {'label': 'businessunit', 'natural': 'businessunit'}),
('locations', {'label': 'locationname', 'natural': 'locationname'}),
('vendors', {'label': 'vendor', 'natural': 'vendor'}),
('models', {'label': 'modelnumber', 'natural': 'modelnumber'}),
('operatingsystems', {'label': 'osname', 'natural': 'osname'}),
('assets', {'label': 'assetnumber', 'natural': 'assetnumber'}),
('computers', {'label': 'hostname', 'natural': 'assetid'}),
('machines', {'label': None, 'natural': 'assetid'}),
])
# Columns a CSV may never set: surrogate keys the database owns, and audit
# stamps. Accepting an id from a spreadsheet is how two sites end up with
# colliding primary keys the first time anyone merges data.
NEVER_IMPORT = {'createddate', 'modifieddate'}
# Credentials are never taken from a file. A CSV that carries password hashes
# gets mailed around, and one that carries plaintext is worse. Users are created
# with a random password and must change it at first login.
PASSWORD_COLUMNS = {'passwordhash', 'password'}
TRUE_VALUES = {'1', 'true', 'yes', 'y', 't'}
FALSE_VALUES = {'0', 'false', 'no', 'n', 'f', ''}
def table_for(name):
table = db.metadata.tables.get(name)
if table is None:
raise ImportError_(
"no such table: %s\nImportable: %s" % (name, ', '.join(IMPORTABLE)))
return table
def assert_schema_ready(tablename):
"""Fail with a sentence, not a traceback, when the schema is not there yet.
Running this against a database that has never been migrated is an easy
mistake - a fresh site, or the wrong DATABASE_URL - and SQLAlchemy's answer
is a wall of stack frames ending in 'no such table'. The operator needs to
be told to run the migrations.
"""
# The SESSION's connection, not db.engine. Taking a separate connection ends
# up rolling back the session's in-flight transaction under a shared-
# connection pool, which silently discarded rows that had just been written.
inspector = sqla_inspect(db.session.connection())
if not inspector.has_table(tablename):
raise ImportError_(
"the '%s' table does not exist in this database.\n"
" Run 'flask db upgrade' first, and check DATABASE_URL points at the\n"
" site you meant." % tablename)
def importable_columns(tablename):
"""Columns an operator may set, in a stable order.
Primary keys are excluded - they are the database's - EXCEPT where the key
is also a foreign key, as on the subtype tables where `assetid` both
identifies the row and points at its asset.
"""
table = table_for(tablename)
primary = {c.name for c in table.primary_key}
out = []
for column in table.columns:
if column.name in NEVER_IMPORT:
continue
if column.name in primary and not column.foreign_keys:
continue
if column.name in PASSWORD_COLUMNS:
continue
out.append(column)
return out
def required_columns(tablename):
primary = {c.name for c in table_for(tablename).primary_key}
return [
c.name for c in importable_columns(tablename)
if not c.nullable and c.name not in primary
and c.default is None and c.server_default is None
]
def _fk_target(column):
"""(tablename, idcolumn) this column points at, or None."""
if not column.foreign_keys:
return None
target = list(column.foreign_keys)[0].column
return target.table.name, target.name
def _label_column(tablename):
"""The human-readable column for a table, used to resolve references to it.
Derived where possible so this does not need maintaining: the schema names
reference tables' label column after the table itself (assettypes.assettype,
locationtypes.locationtype). IMPORTABLE overrides where that rule does not
hold.
"""
spec = IMPORTABLE.get(tablename)
if spec and spec.get('label'):
return spec['label']
table = db.metadata.tables.get(tablename)
if table is None:
return None
singular = tablename[:-1] if tablename.endswith('s') else tablename
for candidate in (singular, singular + 'name', tablename + 'name'):
if candidate in table.columns:
return candidate
return None
def _coerce(column, raw):
"""Turn a CSV string into something the column will accept."""
text = (raw or '').strip()
typename = column.type.__class__.__name__.upper()
if text == '':
return None
if 'BOOL' in typename:
low = text.lower()
if low in TRUE_VALUES:
return True
if low in FALSE_VALUES:
return False
raise ValueError("expected 1 or 0, got '%s'" % text)
if 'INT' in typename:
try:
return int(text)
except ValueError:
raise ValueError("expected a whole number, got '%s'" % text)
if 'FLOAT' in typename or 'NUMERIC' in typename or 'DECIMAL' in typename:
try:
return float(text)
except ValueError:
raise ValueError("expected a number, got '%s'" % text)
if 'DATE' in typename or 'TIME' in typename:
from datetime import datetime
for fmt in ('%Y-%m-%d %H:%M:%S', '%Y-%m-%dT%H:%M:%S', '%Y-%m-%d', '%d/%m/%Y', '%m/%d/%Y'):
try:
return datetime.strptime(text, fmt)
except ValueError:
continue
raise ValueError("expected a date like 2026-08-04, got '%s'" % text)
length = getattr(column.type, 'length', None)
if length and len(text) > length:
raise ValueError("longer than the column allows (%d > %d characters)" % (len(text), length))
return text
class Resolver:
"""Turns names into ids, remembering what it has already looked up.
Rows created earlier in the SAME import are resolvable, so one file can
reference another that was loaded moments ago without a commit in between.
"""
def __init__(self):
self._cache = {}
def resolve(self, tablename, idcolumn, value):
if value is None:
return None
# A number is taken at face value: exports round-trip, and some sites do
# know their ids.
if isinstance(value, int) or (isinstance(value, str) and value.strip().isdigit()):
return int(value)
label = _label_column(tablename)
if not label:
raise ValueError(
"cannot look up %s by name - give the numeric %s instead" % (tablename, idcolumn))
key = (tablename, str(value).strip().lower())
if key in self._cache:
return self._cache[key]
table = db.metadata.tables[tablename]
row = db.session.execute(
table.select().where(
db.func.lower(db.func.trim(table.c[label])) == str(value).strip().lower())
).first()
if row is None:
raise ValueError(
"nothing in %s is named '%s' - add it to %s.csv, or import that file first"
% (tablename, value, tablename))
found = getattr(row, idcolumn)
self._cache[key] = found
return found
def remember(self, tablename, name, rowid):
if name is None:
return
self._cache[(tablename, str(name).strip().lower())] = rowid
class RowProblem(object):
def __init__(self, line, column, message):
self.line = line
self.column = column
self.message = message
def __str__(self):
where = "line %d" % self.line
if self.column:
where += ", column '%s'" % self.column
return "%s: %s" % (where, self.message)
class ImportResult(object):
def __init__(self, tablename):
self.tablename = tablename
self.created = 0
self.updated = 0
self.problems = []
self.skipped = 0
@property
def ok(self):
return not self.problems
def summary(self):
return "%s: %d new, %d updated, %d problem(s)" % (
self.tablename, self.created, self.updated, len(self.problems))
def read_rows(text):
"""Rows as dicts, with the sample/comment lines the templates carry removed."""
reader = csv.reader(io.StringIO(text))
header = None
rows = []
for number, fields in enumerate(reader, 1):
if not fields or all(not f.strip() for f in fields):
continue
if fields[0].lstrip().startswith('#'):
continue
if header is None:
# Strip a UTF-8 BOM if one survived: Excel writes it, and a header
# named '\ufeffassetnumber' would match nothing.
header = [f.strip().lstrip('\ufeff') for f in fields]
continue
# A short row is padded rather than rejected: trailing empty cells are
# routinely dropped by spreadsheet exports.
padded = list(fields) + [''] * (len(header) - len(fields))
rows.append((number, dict(zip(header, padded))))
if header is None:
raise ImportError_('the file has no header row')
return header, rows
def import_csv(tablename, text, resolver=None, commit=False):
"""Validate a CSV against one table and write it into the current transaction.
Validation runs over EVERY row first. Nothing is written unless the whole
file is clean, so a typo on line 400 does not leave 399 rows imported.
Writes always go into the SESSION; this never commits. The caller commits or
rolls back, which is what makes a dry run over a whole folder work: assets
can resolve the locations the previous file just created, because they are
really there inside the transaction, and the rollback removes them again.
Skipping the writes on a dry run instead made every cross-file reference
fail, which is the one thing a folder-wide check exists to verify.
`commit` is retained only for callers that want the old explicit read.
"""
table = table_for(tablename)
spec = IMPORTABLE.get(tablename)
if spec is None:
raise ImportError_(
"%s is not importable.\nImportable: %s" % (tablename, ', '.join(IMPORTABLE)))
assert_schema_ready(tablename)
resolver = resolver or Resolver()
result = ImportResult(tablename)
header, rows = read_rows(text)
allowed = {c.name: c for c in importable_columns(tablename)}
unknown = [h for h in header if h not in allowed]
if unknown:
known = ', '.join(sorted(allowed))
result.problems.append(RowProblem(
1, None, "unknown column(s): %s\n this table accepts: %s" % (', '.join(unknown), known)))
return result
missing = [c for c in required_columns(tablename) if c not in header]
if missing:
result.problems.append(RowProblem(
1, None, "required column(s) missing: %s" % ', '.join(missing)))
return result
natural = spec['natural']
if natural not in header:
result.problems.append(RowProblem(
1, None, "the '%s' column identifies each row and must be present" % natural))
return result
prepared = []
for line, raw in rows:
values = {}
failed = False
for name, column in allowed.items():
if name not in raw:
continue
target = _fk_target(column)
# RESOLVE BEFORE COERCING. A foreign key column is an integer, so
# coercing first rejects every name with "expected a whole number" -
# which defeats the entire point of accepting names. Resolution
# handles the numeric case itself.
if target:
text = (raw[name] or '').strip()
if text == '':
values[name] = None
continue
try:
values[name] = resolver.resolve(target[0], target[1], text)
except ValueError as exc:
result.problems.append(RowProblem(line, name, str(exc)))
failed = True
continue
try:
values[name] = _coerce(column, raw[name])
except ValueError as exc:
result.problems.append(RowProblem(line, name, str(exc)))
failed = True
if failed:
continue
if values.get(natural) in (None, ''):
result.problems.append(RowProblem(line, natural, 'must have a value'))
continue
prepared.append((line, values))
if not result.ok:
return result
# Second pass: decide insert vs update. Counted even on a dry run, because
# "this will change 4000 rows" is the number an operator needs before saying
# yes.
label = _label_column(tablename)
for line, values in prepared:
existing = db.session.execute(
table.select().where(table.c[natural] == values[natural])).first()
if existing is None:
result.created += 1
inserted = db.session.execute(table.insert().values(**values))
newid = inserted.inserted_primary_key[0] if inserted.inserted_primary_key else None
if label and label in values:
resolver.remember(tablename, values[label], newid)
else:
result.updated += 1
db.session.execute(
table.update().where(table.c[natural] == values[natural]).values(**values))
if label and label in values:
primary = list(table.primary_key)[0].name
resolver.remember(tablename, values[label], getattr(existing, primary))
return result
def generate_template(tablename):
"""A header row plus a commented example, built from the live schema.
Generated rather than maintained: a hand-written template set drifts on the
next migration, and silently - the file still looks right.
"""
columns = importable_columns(tablename)
required = set(required_columns(tablename))
header = [c.name for c in columns]
notes = []
for column in columns:
bits = []
bits.append('REQUIRED' if column.name in required else 'optional')
target = _fk_target(column)
if target:
label = _label_column(target[0])
if label:
bits.append("name from %s.csv (or a numeric id)" % target[0])
else:
bits.append("numeric %s id" % target[0])
else:
typename = column.type.__class__.__name__.upper()
if 'BOOL' in typename:
bits.append('1 or 0')
elif 'DATE' in typename or 'TIME' in typename:
bits.append('YYYY-MM-DD')
elif getattr(column.type, 'length', None):
bits.append('text, max %d' % column.type.length)
notes.append("# %-24s %s" % (column.name, ', '.join(bits)))
spec = IMPORTABLE[tablename]
lines = [
"# %s.csv - generated from the ShopDB schema, do not hand-maintain" % tablename,
"#",
"# Rows are matched on '%s'. Re-importing updates a matching row rather" % spec['natural'],
"# than creating a second one, so it is safe to correct this file and run again.",
"#",
"# Columns:",
] + notes + [
"#",
"# Lines starting with # are ignored. Delete the example row before importing.",
','.join(header),
'# ' + ','.join(_example(c, required) for c in columns),
]
return '\n'.join(lines) + '\n'
def _example(column, required):
target = _fk_target(column)
if target:
label = _label_column(target[0])
return ('<%s name>' % target[0][:-1]) if label else '<%s id>' % target[0]
typename = column.type.__class__.__name__.upper()
if 'BOOL' in typename:
return '1'
if 'DATE' in typename or 'TIME' in typename:
return '2026-08-04'
if 'INT' in typename or 'FLOAT' in typename or 'NUMERIC' in typename:
return '0'
return '<%s>' % column.name
def dependency_order(names):
"""The given tables, ordered so a table never precedes what it references."""
known = [t for t in IMPORTABLE if t in set(names)]
return known
def table_from_filename(filename):
stem = re.sub(r'\.csv$', '', filename.strip(), flags=re.I)
stem = stem.rsplit('/', 1)[-1].rsplit('\\', 1)[-1].lower()
return stem

236
tests/test_csv_import.py Normal file
View File

@@ -0,0 +1,236 @@
"""Bulk CSV import: name resolution, all-or-nothing safety, and template truth.
The behaviour under test is what makes a spreadsheet import usable by someone
who does not have the database in front of them. In particular: a foreign key
column accepts the NAME of the thing it points at. Requiring numeric ids means
importing a file, reading back generated ids, and pasting them into the next
file - a workflow nobody finishes.
"""
import pytest
from shopdb.core.services.csvimport import (
IMPORTABLE, ImportError_, Resolver, generate_template, import_csv,
importable_columns, required_columns,
)
from shopdb.extensions import db as _db
REFERENCE = [
('assettypes', 'assettype,description\nMachine,Shop floor\nMeasuring Tool,Gage lab\n'),
('assetstatuses', 'status,description\nActive,In service\nRetired,Out\n'),
('locationtypes', 'locationtype\nBay\nRoom\n'),
]
def seed_reference(resolver):
for table, text in REFERENCE:
result = import_csv(table, text, resolver=resolver, commit=True)
assert result.ok, [str(p) for p in result.problems]
result = import_csv(
'locations',
'locationname,building,locationtypeid\nBay 3,Building 1,Bay\nGage Lab,Building 2,Room\n',
resolver=resolver, commit=True)
assert result.ok, [str(p) for p in result.problems]
def count(table):
return _db.session.execute(_db.text('select count(*) from %s' % table)).scalar()
def test_foreign_keys_resolve_by_name(db):
"""The headline behaviour: write 'Gage Lab', not locationid=2."""
resolver = Resolver()
seed_reference(resolver)
result = import_csv(
'assets',
'assetnumber,name,assettypeid,statusid,locationid\n'
'CMM-01,Zeiss,Measuring Tool,Active,Gage Lab\n',
resolver=resolver, commit=True)
assert result.ok, [str(p) for p in result.problems]
row = _db.session.execute(_db.text(
'select t.assettype, s.status, l.locationname from assets a '
'join assettypes t on t.assettypeid = a.assettypeid '
'join assetstatuses s on s.statusid = a.statusid '
'join locations l on l.locationid = a.locationid')).fetchone()
assert tuple(row) == ('Measuring Tool', 'Active', 'Gage Lab')
def test_numeric_ids_still_work(db):
"""Exports round-trip, and some sites genuinely know their ids."""
resolver = Resolver()
seed_reference(resolver)
typeid = _db.session.execute(_db.text(
"select assettypeid from assettypes where assettype = 'Machine'")).scalar()
result = import_csv(
'assets', 'assetnumber,assettypeid\nMILL-1,%d\n' % typeid,
resolver=resolver, commit=True)
assert result.ok, [str(p) for p in result.problems]
assert count('assets') == 1
def test_unknown_name_names_the_column_and_the_value(db):
"""The error has to be actionable by someone holding a spreadsheet."""
resolver = Resolver()
seed_reference(resolver)
result = import_csv(
'assets',
'assetnumber,assettypeid,locationid\nX-1,Machine,Bay 9\n',
resolver=resolver, commit=False)
assert not result.ok
message = str(result.problems[0])
assert "locationid" in message
assert "Bay 9" in message
assert "locations.csv" in message
def test_reimport_updates_rather_than_duplicating(db):
"""Sites correct their spreadsheet and run it again. That must not double up."""
resolver = Resolver()
seed_reference(resolver)
text = 'assetnumber,name,assettypeid\nCMM-01,Zeiss,Machine\n'
first = import_csv('assets', text, resolver=resolver, commit=True)
assert (first.created, first.updated) == (1, 0)
renamed = 'assetnumber,name,assettypeid\nCMM-01,Zeiss Contura,Machine\n'
second = import_csv('assets', renamed, resolver=resolver, commit=True)
assert (second.created, second.updated) == (0, 1)
assert count('assets') == 1
assert _db.session.execute(_db.text(
"select name from assets where assetnumber = 'CMM-01'")).scalar() == 'Zeiss Contura'
def test_one_bad_row_writes_nothing(db):
"""A typo on the last line must not leave the earlier lines imported."""
resolver = Resolver()
seed_reference(resolver)
result = import_csv(
'assets',
'assetnumber,assettypeid\n'
'GOOD-1,Machine\n'
'GOOD-2,Machine\n'
'BAD-3,NoSuchType\n',
resolver=resolver, commit=True)
assert not result.ok
assert count('assets') == 0, 'rows were written despite a failure'
def test_unknown_column_is_rejected_with_the_accepted_list(db):
"""The failure mode of a hand-written template: invented columns."""
resolver = Resolver()
seed_reference(resolver)
result = import_csv(
'vendors', 'vendor,zipcode\nDell,78682\n', resolver=resolver, commit=False)
assert not result.ok
message = str(result.problems[0])
assert 'zipcode' in message
assert 'this table accepts' in message
def test_missing_required_column_is_rejected(db):
result = import_csv('assets', 'name\nNo asset number\n', commit=False)
assert not result.ok
assert 'assetnumber' in str(result.problems[0])
def test_booleans_take_1_and_0_and_reject_prose(db):
resolver = Resolver()
seed_reference(resolver)
ok = import_csv('assets', 'assetnumber,assettypeid,isactive\nA-1,Machine,0\n',
resolver=resolver, commit=True)
assert ok.ok, [str(p) for p in ok.problems]
bad = import_csv('assets', 'assetnumber,assettypeid,isactive\nA-2,Machine,maybe\n',
resolver=resolver, commit=False)
assert not bad.ok
assert 'isactive' in str(bad.problems[0])
def test_comment_and_blank_lines_are_ignored(db):
"""The templates ship a commented example. Importing one unedited should be
a no-op, not a row of literal placeholder text."""
resolver = Resolver()
seed_reference(resolver)
result = import_csv(
'assets',
'# a comment\nassetnumber,assettypeid\n\n# <assetnumber>,<type>\nREAL-1,Machine\n',
resolver=resolver, commit=True)
assert result.ok, [str(p) for p in result.problems]
assert result.created == 1
def test_passwords_are_never_importable():
"""A CSV carrying password hashes gets mailed around; one carrying plaintext
is worse. Neither column may be set from a file."""
for table in IMPORTABLE:
names = {c.name for c in importable_columns(table)}
assert 'passwordhash' not in names
assert 'password' not in names
def test_generated_templates_only_contain_real_columns(db):
"""The regression that motivated all of this.
A hand-written template set had invented columns on 7 of 11 tables and named
a table that does not exist. Generating from the schema is what makes that
impossible - so assert the generator actually does it.
"""
for table in IMPORTABLE:
text = generate_template(table)
header = [line for line in text.splitlines() if line and not line.startswith('#')][0]
real = {c.name for c in importable_columns(table)}
for column in header.split(','):
assert column in real, '%s.csv offers a column that does not exist: %s' % (table, column)
def test_every_template_includes_its_required_columns(db):
"""A template that omits a required column produces a file that cannot import."""
for table in IMPORTABLE:
header = [line for line in generate_template(table).splitlines()
if line and not line.startswith('#')][0].split(',')
for column in required_columns(table):
assert column in header, '%s.csv is missing required column %s' % (table, column)
def test_importable_tables_all_exist(db):
"""A registry entry naming a table that is not in the schema would fail only
when someone tried to use it."""
for table in IMPORTABLE:
assert table in _db.metadata.tables, '%s is registered but not in the schema' % table
def test_unimportable_table_is_refused(db):
with pytest.raises(ImportError_):
import_csv('auditlog', 'x\n1\n', commit=False)
def test_dry_run_reports_counts_then_rolls_back(db):
"""The number an operator needs before saying yes - and no trace afterwards.
Writes go into the transaction even on a dry run, which is what lets a later
file resolve names an earlier one created. The rollback is what makes it a
dry run.
"""
resolver = Resolver()
seed_reference(resolver)
result = import_csv(
'assets', 'assetnumber,assettypeid\nD-1,Machine\nD-2,Machine\n',
resolver=resolver)
assert result.ok
assert result.created == 2
_db.session.rollback()
assert count('assets') == 0
def test_a_dry_run_over_several_files_resolves_across_them(db):
"""The whole-folder check: assets must resolve locations that only exist
because an earlier file in the same run created them."""
resolver = Resolver()
seed_reference(resolver)
result = import_csv(
'assets',
'assetnumber,assettypeid,locationid\nX-1,Machine,Bay 3\n',
resolver=resolver)
assert result.ok, [str(p) for p in result.problems]
_db.session.rollback()
assert count('assets') == 0