diff --git a/olclient/__init__.py b/olclient/__init__.py
index 239f92bb..6ab49cf8 100644
--- a/olclient/__init__.py
+++ b/olclient/__init__.py
@@ -16,3 +16,13 @@
 from olclient.bots import AbstractBotJob
 from olclient.openlibrary import OpenLibrary
 from olclient.common import Book, Author
+from olclient.imports import (
+    DataProvider,
+    DataProviderRecord,
+    JSONLProvider,
+    OLAuthor,
+    OLContributor,
+    OLImportRecord,
+    OLLink,
+    PaginatedAPIProvider,
+)
diff --git a/olclient/imports.py b/olclient/imports.py
new file mode 100644
index 00000000..6909e6a6
--- /dev/null
+++ b/olclient/imports.py
@@ -0,0 +1,281 @@
+"""
+imports.py
+~~~~~~~~~~
+
+Primitives for mapping external data sources to Open Library import records.
+
+Architecture mirrors pyopds2 (https://github.com/ArchiveLabs/pyopds2):
+
+    DataProvider      — knows how to traverse/iterate a source
+    DataProviderRecord — Pydantic model of one source record; to_ol_import() converts it
+    OLImportRecord    — Pydantic model mirroring olclient/schemata/import.schema.json
+
+Concrete source implementations live in openlibrary-bots/sources/<slug>/.
+
+Usage::
+
+    class MyRecord(DataProviderRecord):
+        title: str
+        authors: list[dict]
+
+        def to_ol_import(self):
+            return OLImportRecord(
+                title=self.title,
+                source_records=["mysource:123"],
+                authors=[OLAuthor(name=a["name"]) for a in self.authors],
+                publishers=["Unknown"],
+                publish_date="2024",
+            )
+
+    class MyProvider(JSONLProvider):
+        SOURCE_SLUG = "mysource"
+        SOURCE_URL = "https://example.com/data.jsonl"
+        RECORD_CLASS = MyRecord
+
+    for record in MyProvider().iter_ol_records():
+        print(record.model_dump(exclude_none=True))
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import re
+import urllib.request
+from abc import ABC, abstractmethod
+from typing import Iterator, List, Literal, Optional
+
+from pydantic import BaseModel, ConfigDict, ValidationError, field_validator
+
+log = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# OL Import Record — mirrors import.schema.json
+# ---------------------------------------------------------------------------
+
+
+class OLLink(BaseModel):
+    model_config = ConfigDict(extra='forbid')
+
+    url: str
+    title: str
+
+
+class OLAuthor(BaseModel):
+    model_config = ConfigDict(extra='forbid')
+
+    name: str
+    personal_name: Optional[str] = None
+    birth_date: Optional[str] = None
+    death_date: Optional[str] = None
+    entity_type: Optional[Literal["person", "org", "event"]] = None
+    title: Optional[str] = None
+
+
+class OLContributor(BaseModel):
+    model_config = ConfigDict(extra='forbid')
+
+    name: str
+    role: Optional[str] = None
+
+
+class OLImportRecord(BaseModel):
+    """
+    Pydantic model mirroring olclient/schemata/import.schema.json.
+
+    extra='forbid' enforces additionalProperties=false at the Python level,
+    catching unknown fields before they reach the OL API (which silently drops them).
+    """
+
+    model_config = ConfigDict(extra='forbid')
+
+    # Required
+    title: str
+    source_records: List[str]
+    authors: List[OLAuthor]
+    publishers: List[str]
+    publish_date: str
+
+    # Optional descriptive
+    subtitle: Optional[str] = None
+    description: Optional[str] = None
+    notes: Optional[str] = None
+    edition_name: Optional[str] = None
+    by_statement: Optional[str] = None
+
+    # Optional publication
+    publish_places: Optional[List[str]] = None
+    publish_country: Optional[str] = None
+
+    # Optional physical
+    number_of_pages: Optional[int] = None
+    pagination: Optional[str] = None
+    physical_format: Optional[str] = None
+    physical_dimensions: Optional[str] = None
+    weight: Optional[str] = None
+
+    # Optional classification
+    lccn: Optional[List[str]] = None
+    oclc_numbers: Optional[List[str]] = None
+    isbn_10: Optional[List[str]] = None
+    isbn_13: Optional[List[str]] = None
+    lc_classifications: Optional[List[str]] = None
+    dewey_decimal_class: Optional[List[str]] = None
+
+    # Optional subjects
+    subjects: Optional[List[str]] = None
+    subject_times: Optional[List[str]] = None
+    subject_people: Optional[List[str]] = None
+    subject_places: Optional[List[str]] = None
+
+    # Optional language
+    languages: Optional[List[str]] = None
+    translated_from: Optional[List[str]] = None
+    translation_of: Optional[str] = None
+
+    # Optional relations
+    series: Optional[List[str]] = None
+    contributions: Optional[List[str]] = None
+    work_titles: Optional[List[str]] = None
+    other_titles: Optional[List[str]] = None
+    contributor: Optional[List[OLContributor]] = None
+    table_of_contents: Optional[list] = None
+
+    # Optional links / cover
+    links: Optional[List[OLLink]] = None
+    cover: Optional[str] = None  # URL for edition's cover image
+
+    # Optional identifiers (external site IDs; values are lists per schema)
+    identifiers: Optional[dict[str, List[str]]] = None
+
+    @field_validator('identifiers')
+    @classmethod
+    def _validate_identifier_keys(cls, v: Optional[dict]) -> Optional[dict]:
+        if v:
+            for key in v:
+                if not re.match(r'^\w+$', key):
+                    raise ValueError(
+                        f"Identifier key {key!r} must match ^\\w+ (letters, digits, underscores only)"
+                    )
+        return v
+
+
+# ---------------------------------------------------------------------------
+# DataProviderRecord — abstract base for one source-native record
+# ---------------------------------------------------------------------------
+
+
+class DataProviderRecord(BaseModel, ABC):
+    """
+    Abstract base for one record modelled in the source's native schema.
+
+    Subclass with Pydantic fields matching the raw source data, then implement
+    to_ol_import() to convert to OLImportRecord. Return None to skip the record.
+
+    extra='allow' because source schemas vary and we only extract what we need.
+    """
+
+    model_config = ConfigDict(extra='allow')
+
+    @abstractmethod
+    def to_ol_import(self) -> Optional[OLImportRecord]:
+        """Return an OLImportRecord, or None to exclude this record from the batch."""
+        ...
+
+
+# ---------------------------------------------------------------------------
+# DataProvider — abstract base for traversing a source
+# ---------------------------------------------------------------------------
+
+
+class DataProvider(ABC):
+    """
+    Abstract base for traversing a data source.
+
+    Class attributes:
+        SOURCE_SLUG  stable prefix for source_records field (e.g. "itan", "bwb")
+        TITLE        human-readable name for this source
+
+    Implement iter_records() with your traversal strategy (file, API, feed, S3, …).
+    """
+
+    SOURCE_SLUG: str
+    TITLE: str = "Unnamed Provider"
+
+    @abstractmethod
+    def iter_records(self) -> Iterator[DataProviderRecord]:
+        """Yield all records from the source in source-native form."""
+        ...
+
+    def iter_ol_records(self) -> Iterator[OLImportRecord]:
+        """Yield OLImportRecords, skipping records where to_ol_import() returns None."""
+        for record in self.iter_records():
+            result = record.to_ol_import()
+            if result is not None:
+                yield result
+
+
+# ---------------------------------------------------------------------------
+# JSONLProvider — mixin for JSONL file sources (local path or remote URL)
+# ---------------------------------------------------------------------------
+
+
+class JSONLProvider(DataProvider, ABC):
+    """
+    Streams a JSONL file from a local filesystem path or a remote URL.
+
+    Subclasses must define:
+        SOURCE_URL    local path or http(s) URL
+        RECORD_CLASS  the DataProviderRecord subclass to validate each line into
+
+    Lines that fail JSON parsing or Pydantic validation are skipped with a warning.
+    """
+
+    SOURCE_URL: str
+    RECORD_CLASS: type[DataProviderRecord]
+
+    def iter_records(self) -> Iterator[DataProviderRecord]:
+        if self.SOURCE_URL.startswith(('http://', 'https://')):
+            yield from self._iter_remote(self.SOURCE_URL)
+        else:
+            yield from self._iter_local(self.SOURCE_URL)
+
+    TIMEOUT: int = 30  # seconds; override on subclass for slow sources
+
+    def _iter_remote(self, url: str) -> Iterator[DataProviderRecord]:
+        with urllib.request.urlopen(url, timeout=self.TIMEOUT) as f:
+            for lineno, raw in enumerate(f, 1):
+                yield from self._parse_line(raw, lineno)
+
+    def _iter_local(self, path: str) -> Iterator[DataProviderRecord]:
+        with open(path, 'rb') as f:
+            for lineno, raw in enumerate(f, 1):
+                yield from self._parse_line(raw, lineno)
+
+    def _parse_line(self, raw: bytes, lineno: int) -> Iterator[DataProviderRecord]:
+        try:
+            data = json.loads(raw)
+        except json.JSONDecodeError as exc:
+            log.warning("Line %d: JSON parse error — %s", lineno, exc)
+            return
+        try:
+            yield self.RECORD_CLASS.model_validate(data)
+        except ValidationError as exc:
+            log.warning("Line %d: validation error — %s", lineno, exc)
+
+
+# ---------------------------------------------------------------------------
+# PaginatedAPIProvider — base for paginated REST API sources
+# ---------------------------------------------------------------------------
+
+
+class PaginatedAPIProvider(DataProvider, ABC):
+    """
+    Skeleton for paginated JSON REST APIs.
+
+    PAGE_SIZE is a convention; subclasses manage their own pagination loop inside
+    iter_records(). See JSONLProvider for a complete mixin example.
+    """
+
+    PAGE_SIZE: int = 100
diff --git a/olclient/openlibrary.py b/olclient/openlibrary.py
index b16ddd3d..fa18ef12 100644
--- a/olclient/openlibrary.py
+++ b/olclient/openlibrary.py
@@ -141,6 +141,146 @@ def delete_many(self, ol_ids: List[str], comment: str) -> Response:
             comment=comment
         )
 
+    def submit_batch(self, records, *, dry_run: bool = False) -> dict:
+        """Submit OLImportRecord instances to POST /import/batch/new.
+
+        Any logged-in user may submit. Records from non-admin accounts land in
+        'needs_review' status and require admin approval before ImportBot picks
+        them up. Admin accounts go straight to 'pending'.
+
+        Args:
+            records: iterable of OLImportRecord (or dicts) to submit.
+            dry_run: if True, serialize to JSONL and return the record count
+                     without making any HTTP request.
+
+        Returns a dict with one of two shapes:
+
+        Success::
+
+            {
+                'success': True,
+                'batch_id': 42,
+                'batch_name': 'batch-abc123',
+                'total_submitted': 67,
+                'total_queued': 67,
+                'total_skipped': 0,
+                'batch_url': 'https://openlibrary.org/import/batch/42',
+            }
+
+        Failure (validation errors)::
+
+            {
+                'success': False,
+                'errors': [{'line': 3, 'message': 'Publication year too old'}],
+            }
+
+        Note: a JSON API endpoint for this would be cleaner; this method
+        parses the HTML response from the form-based /import/batch/new endpoint.
+        """
+        from olclient.imports import OLImportRecord
+
+        lines = []
+        for rec in records:
+            if isinstance(rec, OLImportRecord):
+                lines.append(rec.model_dump_json(exclude_none=True))
+            else:
+                lines.append(json.dumps(rec))
+
+        if dry_run:
+            return {'dry_run': True, 'record_count': len(lines)}
+
+        response = self.session.post(
+            f'{self.base_url}/import/batch/new',
+            data={'batchImportText': '\n'.join(lines)},
+        )
+
+        if response.status_code == 403:
+            raise PermissionError("Not authorized — must be logged in to submit a batch")
+
+        text = response.text
+
+        if 'Import failed' in text:
+            errors = [
+                {'line': int(m.group(1)), 'message': m.group(2).strip()}
+                for m in re.finditer(r'Line (\d+):</strong>\s*([^<]+)', text)
+            ]
+            return {'success': False, 'errors': errors}
+
+        if 'Import results for batch' in text:
+            def _int(pattern):
+                m = re.search(pattern, text)
+                return int(m.group(1)) if m else None
+
+            batch_id = _int(r'Batch #(\d+)')
+            name_m = re.search(r'Batch #\d+ \(([^)]+)\)', text)
+            return {
+                'success': True,
+                'batch_id': batch_id,
+                'batch_name': name_m.group(1) if name_m else None,
+                'total_submitted': _int(r'Records submitted:.*?(\d+)'),
+                'total_queued': _int(r'Total queued:.*?(\d+)'),
+                'total_skipped': _int(r'Total skipped:.*?(\d+)'),
+                'batch_url': f'{self.base_url}/import/batch/{batch_id}' if batch_id else None,
+            }
+
+        return {
+            'success': False,
+            'errors': [{'line': 0, 'message': f'Unexpected response (status {response.status_code})'}],
+        }
+
+    def get_batch(self, batch_id: int) -> dict:
+        """Fetch the current status of a batch import.
+
+        Args:
+            batch_id: the integer batch ID returned by submit_batch().
+
+        Returns::
+
+            {
+                'batch_id': 42,
+                'batch_name': 'batch-abc123',
+                'submitter': 'username',
+                'submit_time': '2026-05-07 00:30:00',
+                'status_counts': {
+                    'pending': 10,
+                    'needs_review': 5,
+                    'imported': 52,
+                    'error': 0,
+                },
+                'batch_url': 'https://openlibrary.org/import/batch/42',
+            }
+
+        Note: parses the HTML from GET /import/batch/{id}; a JSON API
+        endpoint would be preferable and is a candidate for a future OL PR.
+        """
+        response = self.session.get(f'{self.base_url}/import/batch/{batch_id}')
+
+        if response.status_code == 404:
+            raise ValueError(f"Batch {batch_id} not found")
+        if response.status_code == 403:
+            raise PermissionError("Not authorized to view this batch")
+
+        text = response.text
+
+        def _field(label):
+            # Handles both "<p>Label: value</p>" and "<p>Label:</p><p>value</p>"
+            m = re.search(rf'{re.escape(label)}\s*([^<\n]+)', text)
+            return m.group(1).strip() if m else None
+
+        # Parse status counts: "<li>status: N</li>"
+        status_counts = {}
+        for m in re.finditer(r'<li>(\w+):\s*(\d+)', text):
+            status_counts[m.group(1)] = int(m.group(2))
+
+        return {
+            'batch_id': batch_id,
+            'batch_name': _field('Batch Name:'),
+            'submitter': _field('Submitter:'),
+            'submit_time': _field('Submit Time:'),
+            'status_counts': status_counts,
+            'batch_url': f'{self.base_url}/import/batch/{batch_id}',
+        }
+
     err = lambda e: logger.exception("Error retrieving OpenLibrary response: %s", e)
 
     @backoff.on_exception(on_giveup=err, **BACKOFF_KWARGS)  # type: ignore
diff --git a/pyproject.toml b/pyproject.toml
index 72cc7042..f1a4d75d 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -17,17 +17,18 @@ classifiers = [
     "Topic :: Software Development :: Libraries",
     "Programming Language :: Python",
     "Programming Language :: Python :: 3",
-    "Programming Language :: Python :: 3.7",
-    "Programming Language :: Python :: 3.8",
     "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
 ]
+requires-python = ">=3.9"
 dependencies = [
     "backoff==2.2.1",
     "internetarchive==5.7.1",
     "jsonpickle==3.0.2",
     "jsonschema==4.21.1",
+    "pydantic>=2.0,<3",
     "requests[security]==2.31.0",
     "simplejson==3.19.2",
 ]
@@ -67,7 +68,7 @@ extend-select = ["C4", "C9", "PLC", "PLE", "W"]
 ignore = ["E703", "E722", "E731", "F401", "E402"]
 line-length=1501
 show-source = true
-target-version = "py37"
+target-version = "py39"
 
 [tool.ruff.mccabe]
 max-complexity = 33
diff --git a/requirements.txt b/requirements.txt
index 51f138ec..38f362d0 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -2,5 +2,6 @@ backoff==2.2.1
 internetarchive==5.7.1
 jsonpickle==3.0.2
 jsonschema==4.21.1
+pydantic>=2.0
 requests[security]==2.31.0
 simplejson==3.19.2
diff --git a/tests/test_batch_submit.py b/tests/test_batch_submit.py
new file mode 100644
index 00000000..4159f5e7
--- /dev/null
+++ b/tests/test_batch_submit.py
@@ -0,0 +1,237 @@
+"""
+Tests for OpenLibrary.submit_batch() and OpenLibrary.get_batch().
+
+All HTTP calls are mocked — no network required.
+"""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from olclient.imports import OLAuthor, OLImportRecord
+from olclient.openlibrary import OpenLibrary
+
+
+# ---------------------------------------------------------------------------
+# Shared fixtures
+# ---------------------------------------------------------------------------
+
+MINIMAL_RECORD = OLImportRecord(
+    title="Test Book",
+    source_records=["itan_technologies:BOO0001"],
+    authors=[OLAuthor(name="Author One")],
+    publishers=["Test Press"],
+    publish_date="2024",
+)
+
+SUCCESS_HTML = """
+<p>Import results for batch: Batch #42 (batch-abc123def456)</p>
+<p>Records submitted: 3</p>
+<p>Total queued: 3</p>
+<p>Total skipped: 0</p>
+<a href="/import/batch/42">View Batch Status</a>
+"""
+
+ERROR_HTML = """
+<p>Import failed.</p>
+<p>No import will be queued until *every* record validates successfully.</p>
+<p>Validation errors:</p>
+<li><strong>Line 2:</strong> Publication year too old</li>
+<li><strong>Line 3:</strong> Source needs ISBN</li>
+"""
+
+BATCH_STATUS_HTML = """
+<p>Batch ID: 42</p>
+<p>Batch Name: batch-abc123def456</p>
+<p>Submitter: mek</p>
+<p>Submit Time: 2026-05-07 00:30:00</p>
+<h2>Status Summary</h2>
+<ul>
+    <li>pending: 52</li>
+    <li>needs_review: 10</li>
+    <li>imported: 5</li>
+    <li>error: 0</li>
+</ul>
+"""
+
+
+@pytest.fixture
+def ol():
+    with patch('olclient.openlibrary.OpenLibrary.login'):
+        client = OpenLibrary()
+        client.base_url = 'https://openlibrary.org'
+        return client
+
+
+def _mock_response(status_code=200, text=''):
+    resp = MagicMock()
+    resp.status_code = status_code
+    resp.text = text
+    return resp
+
+
+# ---------------------------------------------------------------------------
+# submit_batch — dry_run
+# ---------------------------------------------------------------------------
+
+class TestSubmitBatchDryRun:
+    def test_dry_run_returns_count_no_http(self, ol):
+        ol.session.post = MagicMock()
+        result = ol.submit_batch([MINIMAL_RECORD, MINIMAL_RECORD], dry_run=True)
+        assert result == {'dry_run': True, 'record_count': 2}
+        ol.session.post.assert_not_called()
+
+    def test_dry_run_accepts_dicts(self, ol):
+        raw = {'title': 'X', 'source_records': ['test:1'], 'authors': [{'name': 'A'}],
+                'publishers': ['P'], 'publish_date': '2024'}
+        ol.session.post = MagicMock()
+        result = ol.submit_batch([raw], dry_run=True)
+        assert result['record_count'] == 1
+
+    def test_dry_run_empty_records(self, ol):
+        ol.session.post = MagicMock()
+        result = ol.submit_batch([], dry_run=True)
+        assert result['record_count'] == 0
+
+
+# ---------------------------------------------------------------------------
+# submit_batch — success
+# ---------------------------------------------------------------------------
+
+class TestSubmitBatchSuccess:
+    def test_success_parses_batch_id(self, ol):
+        ol.session.post = MagicMock(return_value=_mock_response(200, SUCCESS_HTML))
+        result = ol.submit_batch([MINIMAL_RECORD])
+        assert result['success'] is True
+        assert result['batch_id'] == 42
+
+    def test_success_parses_batch_name(self, ol):
+        ol.session.post = MagicMock(return_value=_mock_response(200, SUCCESS_HTML))
+        result = ol.submit_batch([MINIMAL_RECORD])
+        assert result['batch_name'] == 'batch-abc123def456'
+
+    def test_success_parses_counts(self, ol):
+        ol.session.post = MagicMock(return_value=_mock_response(200, SUCCESS_HTML))
+        result = ol.submit_batch([MINIMAL_RECORD])
+        assert result['total_submitted'] == 3
+        assert result['total_queued'] == 3
+        assert result['total_skipped'] == 0
+
+    def test_success_builds_batch_url(self, ol):
+        ol.session.post = MagicMock(return_value=_mock_response(200, SUCCESS_HTML))
+        result = ol.submit_batch([MINIMAL_RECORD])
+        assert result['batch_url'] == 'https://openlibrary.org/import/batch/42'
+
+    def test_posts_to_correct_endpoint(self, ol):
+        ol.session.post = MagicMock(return_value=_mock_response(200, SUCCESS_HTML))
+        ol.submit_batch([MINIMAL_RECORD])
+        call_url = ol.session.post.call_args[0][0]
+        assert call_url == 'https://openlibrary.org/import/batch/new'
+
+    def test_posts_jsonl_as_form_field(self, ol):
+        ol.session.post = MagicMock(return_value=_mock_response(200, SUCCESS_HTML))
+        ol.submit_batch([MINIMAL_RECORD])
+        form_data = ol.session.post.call_args[1]['data']
+        assert 'batchImportText' in form_data
+        # Each line should be valid JSON
+        import json
+        for line in form_data['batchImportText'].strip().splitlines():
+            parsed = json.loads(line)
+            assert parsed['title'] == 'Test Book'
+
+    def test_ol_import_record_serialized_without_none(self, ol):
+        ol.session.post = MagicMock(return_value=_mock_response(200, SUCCESS_HTML))
+        ol.submit_batch([MINIMAL_RECORD])
+        import json
+        form_data = ol.session.post.call_args[1]['data']
+        parsed = json.loads(form_data['batchImportText'])
+        assert 'subtitle' not in parsed  # excluded because None
+
+
+# ---------------------------------------------------------------------------
+# submit_batch — errors
+# ---------------------------------------------------------------------------
+
+class TestSubmitBatchErrors:
+    def test_validation_errors_returned(self, ol):
+        ol.session.post = MagicMock(return_value=_mock_response(200, ERROR_HTML))
+        result = ol.submit_batch([MINIMAL_RECORD])
+        assert result['success'] is False
+        assert len(result['errors']) == 2
+
+    def test_error_line_numbers_parsed(self, ol):
+        ol.session.post = MagicMock(return_value=_mock_response(200, ERROR_HTML))
+        result = ol.submit_batch([MINIMAL_RECORD])
+        assert result['errors'][0]['line'] == 2
+        assert result['errors'][1]['line'] == 3
+
+    def test_error_messages_parsed(self, ol):
+        ol.session.post = MagicMock(return_value=_mock_response(200, ERROR_HTML))
+        result = ol.submit_batch([MINIMAL_RECORD])
+        assert 'Publication year too old' in result['errors'][0]['message']
+        assert 'Source needs ISBN' in result['errors'][1]['message']
+
+    def test_403_raises_permission_error(self, ol):
+        ol.session.post = MagicMock(return_value=_mock_response(403, 'Forbidden'))
+        with pytest.raises(PermissionError):
+            ol.submit_batch([MINIMAL_RECORD])
+
+    def test_unexpected_response_returns_failure(self, ol):
+        ol.session.post = MagicMock(return_value=_mock_response(200, '<html>Something went wrong</html>'))
+        result = ol.submit_batch([MINIMAL_RECORD])
+        assert result['success'] is False
+        assert result['errors'][0]['line'] == 0
+
+
+# ---------------------------------------------------------------------------
+# get_batch
+# ---------------------------------------------------------------------------
+
+class TestGetBatch:
+    def test_returns_batch_id(self, ol):
+        ol.session.get = MagicMock(return_value=_mock_response(200, BATCH_STATUS_HTML))
+        result = ol.get_batch(42)
+        assert result['batch_id'] == 42
+
+    def test_returns_status_counts(self, ol):
+        ol.session.get = MagicMock(return_value=_mock_response(200, BATCH_STATUS_HTML))
+        result = ol.get_batch(42)
+        assert result['status_counts']['pending'] == 52
+        assert result['status_counts']['needs_review'] == 10
+        assert result['status_counts']['imported'] == 5
+        assert result['status_counts']['error'] == 0
+
+    def test_returns_batch_url(self, ol):
+        ol.session.get = MagicMock(return_value=_mock_response(200, BATCH_STATUS_HTML))
+        result = ol.get_batch(42)
+        assert result['batch_url'] == 'https://openlibrary.org/import/batch/42'
+
+    def test_gets_correct_endpoint(self, ol):
+        ol.session.get = MagicMock(return_value=_mock_response(200, BATCH_STATUS_HTML))
+        ol.get_batch(42)
+        ol.session.get.assert_called_once_with('https://openlibrary.org/import/batch/42')
+
+    def test_returns_batch_name(self, ol):
+        ol.session.get = MagicMock(return_value=_mock_response(200, BATCH_STATUS_HTML))
+        result = ol.get_batch(42)
+        assert result['batch_name'] == 'batch-abc123def456'
+
+    def test_returns_submitter(self, ol):
+        ol.session.get = MagicMock(return_value=_mock_response(200, BATCH_STATUS_HTML))
+        result = ol.get_batch(42)
+        assert result['submitter'] == 'mek'
+
+    def test_returns_submit_time(self, ol):
+        ol.session.get = MagicMock(return_value=_mock_response(200, BATCH_STATUS_HTML))
+        result = ol.get_batch(42)
+        assert result['submit_time'] == '2026-05-07 00:30:00'
+
+    def test_404_raises_value_error(self, ol):
+        ol.session.get = MagicMock(return_value=_mock_response(404, 'Not Found'))
+        with pytest.raises(ValueError, match="not found"):
+            ol.get_batch(99999)
+
+    def test_403_raises_permission_error(self, ol):
+        ol.session.get = MagicMock(return_value=_mock_response(403, 'Forbidden'))
+        with pytest.raises(PermissionError):
+            ol.get_batch(42)
diff --git a/tests/test_imports.py b/tests/test_imports.py
new file mode 100644
index 00000000..b0aef3f6
--- /dev/null
+++ b/tests/test_imports.py
@@ -0,0 +1,415 @@
+"""
+Tests for olclient/imports.py
+
+Coverage:
+- OLImportRecord: required fields, extra-field rejection, optional fields, nested models
+- OLAuthor / OLContributor / OLLink: field validation
+- DataProviderRecord: concrete subclass, to_ol_import(), skip-via-None
+- DataProvider: iter_ol_records() filters None results
+- JSONLProvider: local JSONL file, bad lines skipped, remote URL path exists
+- Cross-validation: OLImportRecord output validates against import.schema.json
+"""
+
+import json
+import os
+from nturl2path import pathname2url
+from typing import Iterator, Optional
+
+import jsonschema
+import pytest
+from pydantic import ValidationError
+
+from olclient.imports import (
+    DataProvider,
+    DataProviderRecord,
+    JSONLProvider,
+    OLAuthor,
+    OLContributor,
+    OLImportRecord,
+    OLLink,
+)
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+SCHEMA_PATH = os.path.join(
+    os.path.dirname(__file__), '..', 'olclient', 'schemata', 'import.schema.json'
+)
+
+with open(SCHEMA_PATH) as _f:
+    _IMPORT_SCHEMA = json.load(_f)
+
+_RESOLVER = jsonschema.RefResolver('file:' + pathname2url(os.path.abspath(SCHEMA_PATH)), _IMPORT_SCHEMA)
+_VALIDATOR = jsonschema.Draft4Validator(_IMPORT_SCHEMA, resolver=_RESOLVER)
+
+
+def assert_valid_schema(record: OLImportRecord) -> None:
+    """Assert that a record's dict form passes the canonical JSON schema."""
+    data = record.model_dump(exclude_none=True)
+    errors = list(_VALIDATOR.iter_errors(data))
+    assert not errors, f"Schema validation errors: {errors}"
+
+
+def minimal_record(**overrides) -> dict:
+    """Return the minimum valid kwargs for OLImportRecord."""
+    base = {
+        "title": "Test Book",
+        "source_records": ["test:001"],
+        "authors": [{"name": "Author One"}],
+        "publishers": ["Test Press"],
+        "publish_date": "2024",
+    }
+    base.update(overrides)
+    return base
+
+
+# ---------------------------------------------------------------------------
+# OLAuthor
+# ---------------------------------------------------------------------------
+
+
+class TestOLAuthor:
+    def test_name_only(self):
+        a = OLAuthor(name="Joan Miró")
+        assert a.name == "Joan Miró"
+
+    def test_all_fields(self):
+        a = OLAuthor(
+            name="Hardy, G. H.",
+            personal_name="Hardy, G. H.",
+            birth_date="1877",
+            death_date="1947",
+            entity_type="person",
+            title="Prof.",
+        )
+        assert a.entity_type == "person"
+
+    def test_rejects_extra_fields(self):
+        with pytest.raises(ValidationError):
+            OLAuthor(name="X", unknown_field="bad")
+
+
+# ---------------------------------------------------------------------------
+# OLContributor
+# ---------------------------------------------------------------------------
+
+
+class TestOLContributor:
+    def test_name_and_role(self):
+        c = OLContributor(name="Jane Smith", role="Editor")
+        assert c.role == "Editor"
+
+    def test_name_only(self):
+        c = OLContributor(name="Jane Smith")
+        assert c.role is None
+
+    def test_rejects_extra_fields(self):
+        with pytest.raises(ValidationError):
+            OLContributor(name="X", extra="bad")
+
+
+# ---------------------------------------------------------------------------
+# OLLink
+# ---------------------------------------------------------------------------
+
+
+class TestOLLink:
+    def test_valid(self):
+        link = OLLink(url="https://example.com", title="Homepage")
+        assert link.url == "https://example.com"
+
+    def test_rejects_extra_fields(self):
+        with pytest.raises(ValidationError):
+            OLLink(url="https://example.com", title="X", rel="alternate")
+
+
+# ---------------------------------------------------------------------------
+# OLImportRecord — construction
+# ---------------------------------------------------------------------------
+
+
+class TestOLImportRecord:
+    def test_minimal_valid(self):
+        r = OLImportRecord(**minimal_record())
+        assert r.title == "Test Book"
+        assert r.subtitle is None
+
+    def test_rejects_missing_title(self):
+        kwargs = minimal_record()
+        del kwargs["title"]
+        with pytest.raises(ValidationError):
+            OLImportRecord(**kwargs)
+
+    def test_rejects_missing_authors(self):
+        kwargs = minimal_record()
+        del kwargs["authors"]
+        with pytest.raises(ValidationError):
+            OLImportRecord(**kwargs)
+
+    def test_rejects_missing_publishers(self):
+        kwargs = minimal_record()
+        del kwargs["publishers"]
+        with pytest.raises(ValidationError):
+            OLImportRecord(**kwargs)
+
+    def test_rejects_missing_publish_date(self):
+        kwargs = minimal_record()
+        del kwargs["publish_date"]
+        with pytest.raises(ValidationError):
+            OLImportRecord(**kwargs)
+
+    def test_rejects_missing_source_records(self):
+        kwargs = minimal_record()
+        del kwargs["source_records"]
+        with pytest.raises(ValidationError):
+            OLImportRecord(**kwargs)
+
+    def test_rejects_extra_field(self):
+        with pytest.raises(ValidationError):
+            OLImportRecord(**minimal_record(), cover_url="https://example.com/img.jpg")
+
+    def test_cover_field_name_is_cover_not_cover_url(self):
+        r = OLImportRecord(**minimal_record(), cover="https://example.com/img.jpg")
+        assert r.cover == "https://example.com/img.jpg"
+
+    def test_identifiers_values_are_lists(self):
+        r = OLImportRecord(
+            **minimal_record(),
+            identifiers={"itan_id": ["ISIN123"]},
+        )
+        assert r.identifiers["itan_id"] == ["ISIN123"]
+
+    def test_full_optional_fields(self):
+        r = OLImportRecord(
+            **minimal_record(),
+            subtitle="A Subtitle",
+            isbn_13=["9780441569595"],
+            languages=["eng"],
+            subjects=["Fiction", "Science Fiction"],
+            series=["Sprawl Trilogy"],
+            contributor=[{"name": "Ed Smith", "role": "Editor"}],
+            links=[{"url": "https://example.com", "title": "Homepage"}],
+            identifiers={"goodreads": ["12345"]},
+            cover="https://covers.example.com/book.jpg",
+        )
+        assert r.subtitle == "A Subtitle"
+        assert r.isbn_13 == ["9780441569595"]
+        assert r.contributor[0].role == "Editor"
+
+    def test_model_dump_excludes_none(self):
+        r = OLImportRecord(**minimal_record())
+        dumped = r.model_dump(exclude_none=True)
+        assert "subtitle" not in dumped
+        assert "title" in dumped
+
+
+# ---------------------------------------------------------------------------
+# OLImportRecord — cross-validation against canonical JSON schema
+# ---------------------------------------------------------------------------
+
+
+class TestOLImportRecordSchema:
+    def test_minimal_passes_json_schema(self):
+        r = OLImportRecord(**minimal_record())
+        assert_valid_schema(r)
+
+    def test_full_record_passes_json_schema(self):
+        r = OLImportRecord(
+            title="Neuromancer",
+            source_records=["test:neuromancer"],
+            authors=[OLAuthor(name="Gibson, William", birth_date="1948", entity_type="person")],
+            publishers=["Ace Books"],
+            publish_date="1984",
+            isbn_10=["0441569595"],
+            isbn_13=["9780441569595"],
+            lccn=["91174394"],
+            languages=["eng"],
+            subjects=["Science fiction"],
+            cover="https://covers.openlibrary.org/b/id/1234-L.jpg",
+        )
+        assert_valid_schema(r)
+
+    def test_identifiers_pass_json_schema(self):
+        r = OLImportRecord(
+            **minimal_record(),
+            identifiers={"itan_id": ["ISIN001"], "project_gutenberg": ["64317"]},
+        )
+        assert_valid_schema(r)
+
+
+# ---------------------------------------------------------------------------
+# DataProviderRecord — concrete subclass
+# ---------------------------------------------------------------------------
+
+
+class _SampleRecord(DataProviderRecord):
+    """Minimal concrete record for testing."""
+
+    isin: str
+    title: str
+    author_name: str
+    publisher: Optional[str] = None
+    publish_date: Optional[str] = None
+    should_skip: bool = False
+
+    def to_ol_import(self) -> Optional[OLImportRecord]:
+        if self.should_skip:
+            return None
+        return OLImportRecord(
+            title=self.title,
+            source_records=[f"test:{self.isin}"],
+            authors=[OLAuthor(name=self.author_name)],
+            publishers=[self.publisher or "Unknown"],
+            publish_date=self.publish_date or "2024",
+        )
+
+
+class TestDataProviderRecord:
+    def test_to_ol_import_returns_record(self):
+        rec = _SampleRecord(isin="001", title="Book A", author_name="Alice")
+        result = rec.to_ol_import()
+        assert isinstance(result, OLImportRecord)
+        assert result.title == "Book A"
+        assert result.source_records == ["test:001"]
+
+    def test_to_ol_import_returns_none_to_skip(self):
+        rec = _SampleRecord(isin="002", title="Junk", author_name="X", should_skip=True)
+        assert rec.to_ol_import() is None
+
+    def test_extra_source_fields_are_allowed(self):
+        # DataProviderRecord uses extra='allow' — source fields beyond the model are fine
+        rec = _SampleRecord(isin="003", title="T", author_name="A", unknown_field="ok")
+        assert rec.to_ol_import() is not None
+
+    def test_output_passes_json_schema(self):
+        rec = _SampleRecord(isin="004", title="Valid", author_name="Author", publish_date="2023")
+        assert_valid_schema(rec.to_ol_import())
+
+
+# ---------------------------------------------------------------------------
+# DataProvider — concrete subclass and iter_ol_records
+# ---------------------------------------------------------------------------
+
+
+class _SampleProvider(DataProvider):
+    SOURCE_SLUG = "test"
+    TITLE = "Test Provider"
+
+    def __init__(self, records):
+        self._records = records
+
+    def iter_records(self) -> Iterator[DataProviderRecord]:
+        yield from self._records
+
+
+class TestDataProvider:
+    def _make_records(self, n_good=3, n_skip=2):
+        good = [_SampleRecord(isin=f"G{i}", title=f"Good {i}", author_name="A") for i in range(n_good)]
+        skip = [_SampleRecord(isin=f"S{i}", title=f"Skip {i}", author_name="A", should_skip=True) for i in range(n_skip)]
+        return good + skip
+
+    def test_iter_records_yields_all(self):
+        provider = _SampleProvider(self._make_records(3, 2))
+        assert sum(1 for _ in provider.iter_records()) == 5
+
+    def test_iter_ol_records_filters_none(self):
+        provider = _SampleProvider(self._make_records(3, 2))
+        results = list(provider.iter_ol_records())
+        assert len(results) == 3
+        assert all(isinstance(r, OLImportRecord) for r in results)
+
+    def test_iter_ol_records_empty_source(self):
+        provider = _SampleProvider([])
+        assert list(provider.iter_ol_records()) == []
+
+
+# ---------------------------------------------------------------------------
+# JSONLProvider — local file
+# ---------------------------------------------------------------------------
+
+
+class _TestJSONLRecord(DataProviderRecord):
+    id: str
+    title: str
+    author: str
+
+    def to_ol_import(self) -> Optional[OLImportRecord]:
+        if not self.title:
+            return None
+        return OLImportRecord(
+            title=self.title,
+            source_records=[f"jsonltest:{self.id}"],
+            authors=[OLAuthor(name=self.author)],
+            publishers=["JSONL Press"],
+            publish_date="2025",
+        )
+
+
+class _TestJSONLProvider(JSONLProvider):
+    SOURCE_SLUG = "jsonltest"
+    RECORD_CLASS = _TestJSONLRecord
+
+    def __init__(self, path: str):
+        self.SOURCE_URL = path
+
+
+class TestJSONLProvider:
+    def test_reads_valid_jsonl(self, tmp_path):
+        path = str(tmp_path / "data.jsonl")
+        records = [
+            {"id": "1", "title": "Book One", "author": "Alice"},
+            {"id": "2", "title": "Book Two", "author": "Bob"},
+        ]
+        with open(path, 'w') as f:
+            for r in records:
+                f.write(json.dumps(r) + '\n')
+
+        provider = _TestJSONLProvider(path)
+        results = list(provider.iter_ol_records())
+        assert len(results) == 2
+        assert results[0].title == "Book One"
+        assert results[1].source_records == ["jsonltest:2"]
+
+    def test_skips_bad_json_lines(self, tmp_path, caplog):
+        import logging
+        path = str(tmp_path / "bad.jsonl")
+        with open(path, 'w') as f:
+            f.write('{"id": "1", "title": "Good", "author": "A"}\n')
+            f.write('NOT JSON\n')
+            f.write('{"id": "3", "title": "Also Good", "author": "B"}\n')
+
+        provider = _TestJSONLProvider(path)
+        with caplog.at_level(logging.WARNING, logger="olclient.imports"):
+            results = list(provider.iter_ol_records())
+
+        assert len(results) == 2
+        assert any("JSON parse error" in m for m in caplog.messages)
+
+    def test_skips_lines_failing_validation(self, tmp_path, caplog):
+        import logging
+        path = str(tmp_path / "invalid.jsonl")
+        with open(path, 'w') as f:
+            f.write('{"id": "1", "title": "Good", "author": "A"}\n')
+            # Missing required 'title' and 'author' fields for _TestJSONLRecord
+            f.write('{"id": "2"}\n')
+
+        provider = _TestJSONLProvider(path)
+        with caplog.at_level(logging.WARNING, logger="olclient.imports"):
+            results = list(provider.iter_ol_records())
+
+        assert len(results) == 1
+        assert any("validation error" in m for m in caplog.messages)
+
+    def test_empty_file(self, tmp_path):
+        path_obj = tmp_path / "empty.jsonl"
+        path_obj.write_text("")
+        provider = _TestJSONLProvider(str(path_obj))
+        assert list(provider.iter_ol_records()) == []
+
+    def test_output_passes_json_schema(self, tmp_path):
+        path = tmp_path / "schema_check.jsonl"
+        path.write_text(json.dumps({"id": "99", "title": "Schema Test", "author": "Tester"}) + "\n")
+        provider = _TestJSONLProvider(str(path))
+        for record in provider.iter_ol_records():
+            assert_valid_schema(record)