feat(dwc): add CacheTableMeta model and cache table infrastructure

Adds the export-app cache infrastructure:
- CacheTableMeta model + migration tracking build state per (mapping, collection)
- export.models shim: re-exports Caroline's Schemamapping/Exportdataset/
  Exportdatasetextension under PascalCase aliases for use throughout the package
- cache.py: get_cache_table_name, create_cache_table, drop_cache_table,
  _build_single_cache, _execute_and_populate, _infer_column_type, build_cache_tables
- dwca_utils.py: shared sanitize/build helpers used by cache and archive code
- Tests for SchemaMapping, ExportDataSet, ExportDataSetExtension, CacheTableMeta,
  and cache table operations

Fixes #7737. Closes overlap with the cache mechanism part of c381907
on dwc/foundation; remaining cache features (orphan cleanup, signal handlers,
build API, progress callbacks) ship in later atomic PRs.

Author: foozleface

specifyweb/backend/export/cache.py

@@ -0,0 +1,209 @@
+"""Cache table operations for DwC export pipeline."""
+import logging
+import re
+from django.db import connection
+
+from .dwca_utils import sanitize_column_name
+
+logger = logging.getLogger(__name__)
+
+
+def get_cache_table_name(mapping_id, collection_id, prefix='dwc_cache'):
+    """Generate a safe cache table name."""
+    return f'{prefix}_{mapping_id}_{collection_id}'
+
+
+def create_cache_table(table_name, columns):
+    """Create a cache table with the given columns.
+
+    columns: list of (column_name, column_type) tuples.
+    An auto-increment primary key is always added.
+    """
+    safe_name = re.sub(r'[^a-zA-Z0-9_]', '', table_name)
+    col_defs = ', '.join(
+        f'`{re.sub(r"[^a-zA-Z0-9_]", "", name)}` {col_type}'
+        for name, col_type in columns
+    )
+    with connection.cursor() as cursor:
+        cursor.execute(f'DROP TABLE IF EXISTS `{safe_name}`')
+        cursor.execute(
+            f'CREATE TABLE `{safe_name}` ('
+            f'`id` INT AUTO_INCREMENT PRIMARY KEY, {col_defs}'
+            f') ENGINE=InnoDB DEFAULT CHARSET=utf8mb4'
+        )
+    logger.info('Created cache table %s', safe_name)
+
+
+def drop_cache_table(table_name):
+    """Drop a cache table if it exists."""
+    safe_name = re.sub(r'[^a-zA-Z0-9_]', '', table_name)
+    with connection.cursor() as cursor:
+        cursor.execute(f'DROP TABLE IF EXISTS `{safe_name}`')
+    logger.info('Dropped cache table %s', safe_name)
+
+
+def build_cache_tables(export_dataset, user=None, progress_callback=None):
+    """Build cache tables for an ExportDataSet's core mapping and all extensions."""
+    core_mapping = export_dataset.coremapping
+    collection = export_dataset.collection
+
+    _build_single_cache(core_mapping, collection, user=user,
+                        progress_callback=progress_callback)
+
+    for ext in export_dataset.extensions.all().order_by('sortorder').iterator(chunk_size=2000):
+        _build_single_cache(ext.schemamapping, collection,
+                            prefix=f'dwc_cache_ext{ext.sortorder}',
+                            user=user, progress_callback=progress_callback)
+
+
+def _build_single_cache(mapping, collection, prefix='dwc_cache', user=None,
+                        progress_callback=None):
+    """Build a single cache table for one SchemaMapping."""
+    from .models import CacheTableMeta
+    from django.utils import timezone
+
+    table_name = get_cache_table_name(mapping.id, collection.id, prefix)
+
+    meta, _ = CacheTableMeta.objects.update_or_create(
+        schemamapping=mapping,
+        defaults={'tablename': table_name, 'buildstatus': 'building'}
+    )
+
+    try:
+        display_fields = [
+            f for f in mapping.query.fields.order_by('position')
+            if getattr(f, 'term', None)
+        ]
+
+        columns = [
+            (sanitize_column_name(f.term), _infer_column_type(f))
+            for f in display_fields
+        ]
+
+        create_cache_table(table_name, columns)
+
+        rowcount = _execute_and_populate(
+            table_name, mapping, collection, user, progress_callback
+        )
+
+        meta.buildstatus = 'idle'
+        meta.lastbuilt = timezone.now()
+        meta.rowcount = rowcount
+        meta.save()
+
+        logger.info('Cache table %s built with %d rows', table_name, rowcount)
+
+    except Exception:
+        meta.buildstatus = 'error'
+        meta.save()
+        logger.exception('Failed to build cache table %s', table_name)
+        raise
+
+
+def _execute_and_populate(table_name, mapping, collection, user, progress_callback=None):
+    """Execute a mapping's query and INSERT results into the cache table.
+
+    Uses SQLAlchemy build_query() to ensure output matches query_to_csv
+    (date formatting, null replacement, etc.), then batch-INSERTs rows.
+
+    Returns the number of rows inserted.
+    """
+    from specifyweb.backend.stored_queries.execution import (
+        build_query, BuildQueryProps, set_group_concat_max_len,
+        apply_special_post_query_processing,
+    )
+    from specifyweb.backend.stored_queries.queryfield import QueryField
+    from specifyweb.backend.stored_queries.models import session_context
+    from .field_adapter import EphemeralFieldAdapter
+
+    query_obj = mapping.query
+    display_fields = [
+        f for f in query_obj.fields.order_by('position')
+        if getattr(f, 'term', None)
+    ]
+    field_specs = [
+        QueryField.from_spqueryfield(EphemeralFieldAdapter(f, force_display=True))
+        for f in display_fields
+    ]
+
+    safe_name = re.sub(r'[^a-zA-Z0-9_]', '', table_name)
+    col_count = len(display_fields)
+    placeholders = ', '.join(['%s'] * col_count)
+    col_names = ', '.join(
+        f'`{sanitize_column_name(f.term)}`'
+        for f in display_fields
+    )
+    insert_sql = f'INSERT INTO `{safe_name}` ({col_names}) VALUES ({placeholders})'
+
+    total = 0
+    BATCH_SIZE = 2000
+
+    with session_context() as session:
+        set_group_concat_max_len(session.connection())
+        sa_query, _ = build_query(
+            session, collection, user,
+            query_obj.contexttableid,
+            field_specs,
+            BuildQueryProps(
+                replace_nulls=True,
+                date_format_override='%Y-%m-%d',
+            ),
+        )
+        sa_query = apply_special_post_query_processing(
+            sa_query, query_obj.contexttableid, field_specs, collection, user,
+            should_list_query=False,
+        )
+
+        batch = []
+        if isinstance(sa_query, list):
+            iterator = iter(sa_query)
+        else:
+            iterator = sa_query.yield_per(BATCH_SIZE)
+
+        for row in iterator:
+            batch.append(tuple(
+                str(v) if v is not None else '' for v in row[1:]
+            ))
+
+            if len(batch) >= BATCH_SIZE:
+                with connection.cursor() as cursor:
+                    cursor.executemany(insert_sql, batch)
+                total += len(batch)
+                batch = []
+                if progress_callback:
+                    progress_callback(total, None)
+
+        if batch:
+            with connection.cursor() as cursor:
+                cursor.executemany(insert_sql, batch)
+            total += len(batch)
+
+    if progress_callback:
+        progress_callback(total, total)
+
+    return total
+
+
+def _infer_column_type(spqueryfield):
+    """Infer a MySQL column type from a Specify query field."""
+    fname = (spqueryfield.fieldname or '').lower()
+
+    if 'guid' in fname or 'uuid' in fname:
+        return 'VARCHAR(256)'
+    if fname in ('id', 'rankid', 'number1', 'number2', 'countamt',
+                 'sortorder', 'position', 'version'):
+        return 'INT'
+    if 'numericyear' in fname or 'numericmonth' in fname or 'numericday' in fname:
+        return 'INT'
+    if fname in ('latitude1', 'latitude2', 'longitude1', 'longitude2',
+                 'latlongaccuracy', 'maxelevation', 'minelevation'):
+        return 'DECIMAL(12,6)'
+    if fname in ('startdate', 'enddate', 'determineddate', 'catalogeddate',
+                 'timestampcreated', 'timestampmodified'):
+        return 'VARCHAR(32)'
+    if fname.startswith('is') or fname.startswith('yes'):
+        return 'VARCHAR(8)'
+    if fname in ('catalognumber', 'altcatalognumber', 'barcode', 'fieldnumber',
+                 'code', 'abbreviation', 'datum'):
+        return 'VARCHAR(256)'
+    return 'TEXT'

specifyweb/backend/export/dwca_utils.py

@@ -0,0 +1,131 @@
+"""Shared utilities for DwC archive generation."""
+import re
+from datetime import date
+from uuid import uuid4
+from xml.etree import ElementTree as ET
+
+
+def sanitize_term_name(term_iri):
+    """Extract the short name from a DwC term IRI.
+
+    'http://rs.tdwg.org/dwc/terms/catalogNumber' -> 'catalogNumber'
+    'http://purl.org/dc/terms/type' -> 'type'
+    """
+    if '/' in term_iri:
+        term_iri = term_iri.rsplit('/', 1)[-1]
+    if '#' in term_iri:
+        term_iri = term_iri.rsplit('#', 1)[-1]
+    return term_iri
+
+
+def sanitize_column_name(name):
+    """Sanitize a term IRI into a valid MySQL column name."""
+    name = sanitize_term_name(name)
+    name = re.sub(r'[^a-zA-Z0-9_]', '_', name)
+    return name[:64]
+
+
+# Known extension rowType URIs
+EXTENSION_ROW_TYPES = {
+    'MeasurementOrFact': 'http://rs.iobis.org/obis/terms/ExtendedMeasurementOrFact',
+    'ResourceRelationship': 'http://rs.tdwg.org/dwc/terms/ResourceRelationship',
+    'Identification': 'http://rs.tdwg.org/dwc/terms/Identification',
+    'Multimedia': 'http://rs.gbif.org/terms/1.0/Multimedia',
+}
+
+
+def build_meta_xml(core_terms, ext_info_list):
+    """Build meta.xml describing the DwC archive structure.
+
+    core_terms: list of full term IRIs for the core file
+    ext_info_list: list of dicts with 'filename' and 'terms' (full IRIs)
+    """
+    archive = ET.Element('archive')
+    archive.set('xmlns', 'http://rs.tdwg.org/dwc/text/')
+    archive.set('metadata', 'eml.xml')
+
+    # Core
+    core = ET.SubElement(archive, 'core')
+    core.set('encoding', 'UTF-8')
+    core.set('fieldsTerminatedBy', ',')
+    core.set('linesTerminatedBy', '\\n')
+    core.set('fieldsEnclosedBy', '"')
+    core.set('ignoreHeaderLines', '1')
+    core.set('rowType', 'http://rs.tdwg.org/dwc/terms/Occurrence')
+
+    files = ET.SubElement(core, 'files')
+    location = ET.SubElement(files, 'location')
+    location.text = 'occurrence.csv'
+
+    if core_terms:
+        id_elem = ET.SubElement(core, 'id')
+        id_elem.set('index', '0')
+
+    for idx, term_iri in enumerate(core_terms):
+        f = ET.SubElement(core, 'field')
+        f.set('index', str(idx))
+        f.set('term', term_iri)
+
+    # Extensions
+    for ext in ext_info_list:
+        extension = ET.SubElement(archive, 'extension')
+        extension.set('encoding', 'UTF-8')
+        extension.set('fieldsTerminatedBy', ',')
+        extension.set('linesTerminatedBy', '\\n')
+        extension.set('fieldsEnclosedBy', '"')
+        extension.set('ignoreHeaderLines', '1')
+        row_type = ext.get('rowType', 'http://rs.tdwg.org/dwc/terms/MeasurementOrFact')
+        extension.set('rowType', row_type)
+
+        files = ET.SubElement(extension, 'files')
+        location = ET.SubElement(files, 'location')
+        location.text = ext['filename']
+
+        coreid = ET.SubElement(extension, 'coreid')
+        coreid.set('index', '0')
+
+        for idx, term_iri in enumerate(ext['terms']):
+            f = ET.SubElement(extension, 'field')
+            f.set('index', str(idx))
+            f.set('term', term_iri)
+
+    return ET.tostring(archive, encoding='unicode', xml_declaration=True)
+
+
+def build_eml_xml(export_dataset):
+    """Build EML metadata. Returns custom EML if uploaded, else generates minimal EML."""
+    if export_dataset.metadata:
+        try:
+            from specifyweb.specify.models import Spappresourcedata
+            data = Spappresourcedata.objects.filter(
+                spappresource=export_dataset.metadata
+            ).first()
+            if data and data.data:
+                content = data.data
+                if isinstance(content, bytes):
+                    content = content.decode('utf-8')
+                return content
+        except Exception:
+            pass
+
+    eml = ET.Element('eml:eml')
+    eml.set('xmlns:eml', 'eml://ecoinformatics.org/eml-2.1.1')
+    eml.set('packageId', str(uuid4()))
+    eml.set('system', 'http://specify.org')
+
+    dataset = ET.SubElement(eml, 'dataset')
+    title = ET.SubElement(dataset, 'title')
+    title.text = export_dataset.exportname
+
+    creator = ET.SubElement(dataset, 'creator')
+    org = ET.SubElement(creator, 'organizationName')
+    org.text = 'Specify Collection'
+
+    pubdate = ET.SubElement(dataset, 'pubDate')
+    pubdate.text = date.today().strftime('%Y-%m-%d')
+
+    abstract = ET.SubElement(dataset, 'abstract')
+    para = ET.SubElement(abstract, 'para')
+    para.text = f'Darwin Core Archive export: {export_dataset.exportname}'
+
+    return ET.tostring(eml, encoding='unicode', xml_declaration=True)

specifyweb/backend/export/migrations/0001_cachetablemeta.py

@@ -0,0 +1,45 @@
+from django.db import migrations, models
+import django.db.models.deletion
+import django.utils.timezone
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        ('specify', '0048_extensions_and_vocabulary'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='CacheTableMeta',
+            fields=[
+                ('id', models.AutoField(db_column='CacheTableMetaID', primary_key=True, serialize=False)),
+                ('tablename', models.CharField(db_column='TableName', max_length=128, unique=True)),
+                ('lastbuilt', models.DateTimeField(blank=True, db_column='LastBuilt', null=True)),
+                ('rowcount', models.IntegerField(blank=True, db_column='RowCount', null=True)),
+                ('buildstatus', models.CharField(
+                    choices=[('idle', 'idle'), ('building', 'building'), ('error', 'error')],
+                    db_column='BuildStatus', default='idle', max_length=16,
+                )),
+                ('builderror', models.TextField(blank=True, db_column='BuildError', null=True)),
+                ('timestampcreated', models.DateTimeField(db_column='TimestampCreated', default=django.utils.timezone.now)),
+                ('timestampmodified', models.DateTimeField(db_column='TimestampModified', default=django.utils.timezone.now)),
+                ('collection', models.ForeignKey(
+                    db_column='CollectionID',
+                    on_delete=django.db.models.deletion.CASCADE,
+                    related_name='+', to='specify.collection',
+                )),
+                ('schemamapping', models.ForeignKey(
+                    db_column='SchemaMappingID',
+                    on_delete=django.db.models.deletion.CASCADE,
+                    related_name='cachetablemetas', to='specify.schemamapping',
+                )),
+            ],
+            options={
+                'db_table': 'cachetablemeta',
+                'indexes': [models.Index(fields=['schemamapping', 'collection'], name='CacheMetaMappingColIDX')],
+            },
+        ),
+    ]