Craft Easy File Import — Specification¶
Version: 1.0 Date: 2026-03-28 Related: specification.md, financial-ecosystem-specification.md
Contents¶
- Vision
- Architecture
- Source connectors
- File parsers
- Import rules and mapping
- Sync modes
- Payment file import and reconciliation
- Data models
- Error handling and correction
- Flows and examples
- Implementation plan
1. Vision¶
A file-based integration layer for tenants that can't or won't build API integrations. Poll SFTP servers, receive file uploads, parse XML/JSON/positional files, and push data to any Craft Easy API endpoint — with configurable sync rules.
External system → File (SFTP / upload) → Craft Easy File Import → API endpoints
│
Parse + Map + Validate
│
POST / PUT / PATCH / DELETE
to any endpoint
Use cases: - Import customer data from legacy system (CSV/XML export) - Receive payment files from bank (positional format) - Sync product catalog from ERP (JSON via SFTP) - Nightly data sync from partner system
Split between packages¶
craft-easy-api (always installed):
├── Settings: FILE_IMPORT_ENABLED = true/false
├── Models: ImportConfiguration, ImportRun, RejectedRecord, UnmatchedPayment
├── Routes: /file-import/* (upload, rejected records, run history)
└── Registered only if FILE_IMPORT_ENABLED = true
craft-easy-file-import (installed when needed):
├── Parsers: FlatFile, CSV, XML (XSLT), JSON (jq)
├── Sync engine: create/upsert/sync/append
├── SFTP connector + polling
└── Runs as a service or via craft-easy-jobs
The API endpoints live in craft-easy-api because: - Admin app needs them (view runs, correct rejected records) - File upload endpoint is an API route - Models/collections must exist regardless of which service writes to them
The processing engine lives in craft-easy-file-import because: - Heavy dependencies (SFTP, XSLT, jq) - Runs as background service, not request-response - Different resource requirements (more memory, longer timeout)
2. Architecture¶
┌──────────────────────────────────────────────────────────────┐
│ craft-easy-file-import │
│ │
│ ┌─ Source Connectors ─────────────────────────────────────┐ │
│ │ SFTP Poller │ File Upload API │ Watched Directory │ │
│ └───────┬───────────────┬────────────────────┬────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌─ File Detection ────────────────────────────────────────┐ │
│ │ Detect file type (XML, JSON, CSV, positional) │ │
│ │ Match to import configuration │ │
│ └──────────────────────┬──────────────────────────────────┘ │
│ │ │
│ ┌─ Parser ─────────────▼──────────────────────────────────┐ │
│ │ XML Parser │ JSON Parser │ CSV Parser │ Position │ │
│ └──────────────────────┬──────────────────────────────────┘ │
│ │ Parsed records │
│ ┌─ Mapper + Validator ─▼──────────────────────────────────┐ │
│ │ Apply field mapping (source field → API field) │ │
│ │ Apply transformations (date formats, lookups, etc.) │ │
│ │ Validate against API schema │ │
│ └──────────────────────┬──────────────────────────────────┘ │
│ │ Validated records │
│ ┌─ Sync Engine ────────▼──────────────────────────────────┐ │
│ │ Apply sync rules (create / update / upsert / delete) │ │
│ │ POST / PUT / PATCH / DELETE to API endpoint │ │
│ │ Track results (created, updated, skipped, failed) │ │
│ └──────────────────────┬──────────────────────────────────┘ │
│ │ │
│ ┌─ Result Log ─────────▼──────────────────────────────────┐ │
│ │ Import run history, per-record status, error details │ │
│ └─────────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────────┐
│ craft-easy-api (endpoints) │
└──────────────────────────────────────────────────────────────┘
3. Source Connectors¶
All source connectors are managed via API endpoints so everything can be configured through the admin UI.
3.1 SFTP Connections (API-managed)¶
SFTP connections are their own resource — one connection can be used by multiple import configurations.
API endpoints (in craft-easy-api, requires FILE_IMPORT_ENABLED):
GET /file-import/sftp-connections — List all
POST /file-import/sftp-connections — Create
GET /file-import/sftp-connections/{id} — Get
PATCH /file-import/sftp-connections/{id} — Update
DELETE /file-import/sftp-connections/{id} — Delete
POST /file-import/sftp-connections/{id}/test — Test connection
class SFTPConnection(BaseDocument):
"""SFTP server connection — managed via API, used by import configurations."""
tenant_id: PydanticObjectId
name: str # "Bank payment server"
is_enabled: bool = True
# Connection
host: str
port: int = 22
username: str
auth_type: str # "password" | "key"
password: Optional[str] = Field(default=None, json_schema_extra={"encrypted": True})
private_key: Optional[str] = Field(default=None, json_schema_extra={"encrypted": True})
host_key_fingerprint: Optional[str] = None # For strict host verification
# Status (readonly, updated by polling service)
last_connected_at: Optional[datetime] = Field(default=None, json_schema_extra={"readonly": True})
last_connection_error: Optional[str] = Field(default=None, json_schema_extra={"readonly": True})
class TenantConfig:
tenant_scoped = True
class Settings:
name = "sftp_connections"
3.2 SFTP Polling Configuration¶
Polling rules are part of the import configuration — referencing an SFTP connection:
class SFTPPollingConfig(BaseModel):
"""How to poll an SFTP server for files."""
sftp_connection_id: PydanticObjectId # References SFTPConnection
remote_directory: str # "/outgoing/payments/"
file_pattern: str # "*.xml" | "PAYMENT_*.csv" | "*.dat"
poll_schedule: str # Cron: "*/15 * * * *" (every 15 min)
# After processing
after_import: str # "move" | "delete" | "leave"
archive_directory: Optional[str] = None # For "move": "/archive/2026/03/"
3.3 File Upload API¶
Direct upload via API endpoint:
POST /file-import/upload
Content-Type: multipart/form-data
file: <binary>
config_id: <import configuration ID>
3.4 Complete API endpoint overview¶
All endpoints in craft-easy-api (enabled via FILE_IMPORT_ENABLED=true):
SFTP Connections:
GET /file-import/sftp-connections — List
POST /file-import/sftp-connections — Create
GET /file-import/sftp-connections/{id} — Get
PATCH /file-import/sftp-connections/{id} — Update
DELETE /file-import/sftp-connections/{id} — Delete
POST /file-import/sftp-connections/{id}/test — Test connection
Import Configurations:
GET /file-import/configurations — List
POST /file-import/configurations — Create
GET /file-import/configurations/{id} — Get
PATCH /file-import/configurations/{id} — Update
DELETE /file-import/configurations/{id} — Delete
File Upload:
POST /file-import/upload — Upload file for import
Import Runs:
GET /file-import/runs — List runs (with status filter)
GET /file-import/runs/{id} — Get run detail + summary
Rejected Records:
GET /file-import/rejected — List rejected (filter by run/status)
GET /file-import/rejected/{id} — Get single (full detail + form errors)
PATCH /file-import/rejected/{id} — Correct data or skip
POST /file-import/rejected/retry-all — Retry all corrected
POST /file-import/rejected/skip-all — Skip all remaining
Unmatched Payments:
GET /file-import/unmatched-payments — List unmatched
GET /file-import/unmatched-payments/{id} — Get detail
PATCH /file-import/unmatched-payments/{id} — Match manually or write off
3.3 Watched Directory (local/mounted)¶
For containerized deployments with mounted volumes:
class DirectorySource(BaseModel):
path: str # "/import/incoming/"
file_pattern: str
poll_interval_seconds: int = 60
4. File Parsers¶
4.1 Supported formats¶
| Format | Extension | Description | Example |
|---|---|---|---|
| Flat file | .dat, .txt | Fixed-width positional with hierarchical post types | Swedish bank files (BGC, Autogiro), legacy systems |
| CSV | .csv | Delimited with full dialect configuration | Spreadsheet exports, ERP exports |
| XML | .xml | Transform via XSLT | Bank files (ISO 20022), ERP exports |
| JSON | .json | Transform via jq expressions | API dumps, modern systems |
| JSON Lines | .jsonl | One JSON object per line | Streaming exports |
4.2 Flat file parser (positional)¶
The most complex parser — handles fixed-width positional files with hierarchical post types (parent-child relationships between record types). Designed for Swedish bank file formats (Bankgirot, Autogiro, Kronofogden) and similar legacy formats.
All positions are 1-indexed (matching how file format specifications are documented).
class FlatFileConfig(BaseModel):
"""Configuration for parsing fixed-width positional files."""
encoding: str = "latin-1" # Common for Swedish bank files
post_types: list[PostType]
class PostType(BaseModel):
"""A record/post type identified by its prefix."""
prefix: str # "01", "TK20", "TL10" — matched against start of each line
length: int # Total length of the post (number of characters per line)
fields: list[PostField]
# Hierarchical: child posts that belong to this post type
child_prefixes: list[str] = []
# If present, posts with listed prefixes that come after this line (without
# other lines in between) are added as children. Enables tree structures.
# E.g. PostType "20" with child_prefixes ["25", "26"] means TK25 and TK26
# lines following a TK20 line are children of that TK20 record.
class PostField(BaseModel):
"""A field at a fixed position in the line."""
name: str # Field name — use names from format specification for traceability
start_position: int # 1-indexed start (prefix included in index)
end_position: int # 1-indexed, exclusive. Field "A1 Hello" with start=3, end=8 → "Hello"
type: str # "string" | "integer" | "float" | "date"
# String options
string_format: str = "adjust_right" # "adjust_right" | "adjust_left"
# adjust_right: filler on the left (right-aligned), stripped from left
# adjust_left: filler on the right (left-aligned), stripped from right
filler: Optional[str] = None
# Default: space for strings, "0" for integers/floats
# Filler is stripped based on string_format direction
# Float options
float_decimals: int = 2 # Treat last N characters as decimals: "0000029900" → 299.00
float_separator: Optional[str] = None # If present (e.g. "."), use instead of float_decimals
# Date options
date_format: str = "date"
# Shortcuts: "date" = YYYYMMDD, "time" = HHmmss, "datetime" = YYYYMMDDHHmmss
# Or custom strftime format: "%y%m%d" for YYMMDD
# Default value if field is empty/all filler
default: Optional[Any] = None
Hierarchical post types (child_prefixes)¶
Many file formats have parent-child relationships between record types. For example, a Bankgirot file where each payment (TK20) can have multiple detail records (TK25, TK26):
TK01 HEADER ← Root: header
TK20 Payment 1 ← Root: payment
TK25 Detail for payment 1 ← Child of TK20
TK26 Extra info for payment 1 ← Child of TK20
TK20 Payment 2 ← Root: new payment
TK25 Detail for payment 2 ← Child of TK20
TK99 FOOTER ← Root: footer
Configuration:
post_types=[
PostType(prefix="TK20", length=80, child_prefixes=["TK25", "TK26"], fields=[...]),
PostType(prefix="TK25", length=80, fields=[...]),
PostType(prefix="TK26", length=80, fields=[...]),
]
Parsed result — tree structure:
[
{"type": "TK20", "fields": {"amount": 299.00}, "children": [
{"type": "TK25", "fields": {"detail": "OCR reference"}},
{"type": "TK26", "fields": {"info": "Customer name"}}
]},
{"type": "TK20", "fields": {"amount": 499.00}, "children": [
{"type": "TK25", "fields": {"detail": "OCR reference"}}
]}
]
Nesting can go deeper — a child post type can also have child_prefixes, creating multi-level trees.
Example: Full Bankgirot configuration
bankgirot_config = FlatFileConfig(
encoding="latin-1",
post_types=[
PostType(
prefix="01",
length=80,
fields=[
PostField(name="sender", start_position=3, end_position=12, type="string"),
PostField(name="date", start_position=13, end_position=20, type="date"),
],
),
PostType(
prefix="20",
length=80,
child_prefixes=["25", "26"],
fields=[
PostField(name="bankgiro", start_position=3, end_position=12, type="string"),
PostField(name="reference", start_position=13, end_position=37, type="string"),
PostField(name="amount", start_position=38, end_position=55, type="float",
float_decimals=2, filler="0"),
PostField(name="currency", start_position=56, end_position=58, type="string"),
PostField(name="payment_date", start_position=59, end_position=66, type="date"),
],
),
PostType(
prefix="25",
length=80,
fields=[
PostField(name="info_text", start_position=3, end_position=77, type="string"),
],
),
PostType(
prefix="99",
length=80,
fields=[
PostField(name="record_count", start_position=3, end_position=10, type="integer"),
PostField(name="total_amount", start_position=11, end_position=28, type="float",
float_decimals=2),
],
),
],
)
4.3 CSV parser¶
Full CSV dialect configuration, matching Python's csv module semantics:
class CSVParserConfig(BaseModel):
"""CSV parser with full dialect support."""
encoding: str = "utf-8"
# Dialect (how to parse the file)
dialect: CSVDialect
# Column names — omit if file has header row (inferred automatically)
columns: Optional[list[str]] = None
# Per-column type coercions
coercions: list[CSVCoercion] = []
class CSVDialect(BaseModel):
"""CSV dialect — mirrors Python csv module."""
delimiter: str = ";" # "semicolon" | "comma" | "tab" or any single character
quotechar: Optional[str] = '"'
escapechar: Optional[str] = None
quoting: str = "QUOTE_MINIMAL"
# "QUOTE_ALL" | "QUOTE_MINIMAL" | "QUOTE_NONNUMERIC" | "QUOTE_NONE" |
# "QUOTE_NOTNULL" | "QUOTE_STRINGS"
skipinitialspace: bool = False
strict: bool = False
class CSVCoercion(BaseModel):
"""Type coercion for a specific CSV column."""
column: str # Column name (from header or columns list)
type: str # "integer" | "float" | "date"
date_format: Optional[str] = None # "date" | "time" | "datetime" | custom strftime
float_separator: Optional[str] = None # "." or "," — defaults to "."
4.4 XML parser (XSLT)¶
Transform XML using XSLT rather than XPath mapping. More powerful — can handle any XML structure:
class XMLParserConfig(BaseModel):
"""XML parser using XSLT transformation."""
encoding: str = "utf-8"
xslt: str # XSLT stylesheet as string
# The XSLT should transform the XML into a flat list of records
# that the mapper can process.
XSLT approach is more powerful than XPath mapping because: - Handles namespaces naturally - Can restructure deeply nested XML - Can filter, sort, and compute values during transformation - Standard — XSLT expertise is widely available
For simple cases, a helper generates XSLT from XPath config:
class XMLSimpleConfig(BaseModel):
"""Simplified XML config — auto-generates XSLT internally."""
root_xpath: str # "/Payments/Payment"
namespace: Optional[dict[str, str]] = None
field_mappings: dict[str, str] # target_field → XPath
4.5 JSON parser (jq)¶
Transform JSON using jq expressions. Handles any JSON structure:
class JSONParserConfig(BaseModel):
"""JSON parser using jq transformation."""
jq: str # jq expression
# The jq expression should output a list of flat objects.
# Example: ".data.transactions[] | {amount: .sum, ref: .reference}"
jq is the standard for JSON transformation: - Handles nested structures, arrays, filtering - Widely known (used in shell scripting, APIs) - Single expression can do complex transformations
For simple cases (flat JSON arrays), jq defaults to .[] (identity — each array item is a record).
5. Import Rules and Mapping¶
5.1 Import configuration¶
Each tenant configures how files map to API endpoints:
class ImportConfiguration(BaseDocument):
"""Defines how a file type maps to an API endpoint."""
tenant_id: PydanticObjectId
name: str # "Bankgirot payments"
is_enabled: bool = True
# Source
source_type: str # "sftp" | "upload" | "directory"
source_config: dict # SFTPSource, etc.
# Parser
file_format: str # "xml" | "json" | "jsonl" | "csv" | "positional"
parser_config: dict # Format-specific config
# Target endpoint
target_endpoint: str # "/payments" | "/customers"
# Sync rules
sync_mode: str # "create" | "upsert" | "sync" | "append"
sync_key: Optional[str] = None # Field used for matching: "reference", "external_id"
# Field mapping
field_mapping: list[FieldMapping]
# Validation
skip_invalid_records: bool = True # False = abort entire file on first error
# Hooks
pre_import_job: Optional[str] = None # craft-easy-jobs job to run before
post_import_job: Optional[str] = None # Job to run after (e.g. reconciliation)
class TenantConfig:
tenant_scoped = True
class Settings:
name = "import_configurations"
class FieldMapping(BaseModel):
"""Maps a source field to an API field."""
source_field: str # Field name from parser output
target_field: str # API field name
transform: Optional[str] = None # "uppercase" | "lowercase" | "trim" | "date:YYYYMMDD"
default_value: Optional[str] = None # If source is empty
required: bool = False
lookup: Optional[FieldLookup] = None # Look up value from another endpoint
class FieldLookup(BaseModel):
"""Look up a value from another endpoint during mapping."""
endpoint: str # "/customers"
match_field: str # "external_id"
return_field: str # "_id"
# Example: source has customer_number "C001" → look up /customers?external_id=C001 → get _id
6. Sync Modes¶
| Mode | HTTP Method | Behavior |
|---|---|---|
| create | POST | Always create new records. Fail on duplicate. |
| upsert | POST or PATCH | Create if not exists (by sync_key), update if exists. |
| sync | POST, PATCH, DELETE | Full sync — create new, update existing, delete records not in file. |
| append | POST | Always create, even if duplicate sync_key exists (e.g. payment transactions). |
Upsert logic¶
for record in parsed_records:
existing = await api.get(f"{endpoint}?{sync_key}={record[sync_key]}")
if existing:
# PATCH with changes only
await api.patch(f"{endpoint}/{existing['_id']}", changed_fields)
else:
# POST new record
await api.post(endpoint, record)
Sync logic (full sync)¶
file_keys = {record[sync_key] for record in parsed_records}
existing = await api.get(f"{endpoint}?per_page=10000")
existing_keys = {item[sync_key] for item in existing['items']}
to_create = file_keys - existing_keys # In file but not in API
to_update = file_keys & existing_keys # In both
to_delete = existing_keys - file_keys # In API but not in file
7. Payment File Import and Reconciliation¶
7.1 Payment files as a special case¶
Bank payment files (Bankgirot, SEPA, etc.) need special handling because received payments must be: 1. Matched to outstanding claims/invoices 2. Applied according to the settlement order (avräkningsordning) 3. Recorded in bookkeeping 4. Included in revenue split calculations
7.2 Payment import flow¶
Bank payment file (SFTP)
│
▼
Parse file → list of payments
│
▼
For each payment:
│
├─ Match to claim (by reference/OCR number)
│ ├─ Found → Apply payment via SettlementOrderService
│ │ → Create journal entries
│ │ → Calculate revenue split
│ │ → Update claim status
│ └─ Not found → Mark as unmatched → manual review
│
▼
Import run complete
│
├─ Summary: 45 matched, 3 unmatched, 0 errors
└─ Unmatched payments → notification to tenant admin
7.3 Payment import configuration¶
class PaymentImportConfig(BaseModel):
"""Special config for payment file imports."""
reference_field: str # Which parsed field contains the payment reference/OCR
amount_field: str # Which parsed field contains the amount
date_field: str # Payment date
currency_field: Optional[str] = None # Or fixed currency below
fixed_currency: Optional[str] = "SEK"
# Matching
match_claim_by: str # "payment_reference" | "invoice_number" | "customer_number"
# When payment doesn't match any claim
unmatched_action: str # "skip" | "create_unmatched_record" | "fail"
# Auto-reconciliation
auto_reconcile: bool = True # Apply payment immediately using settlement order
create_journal_entries: bool = True # Generate bookkeeping entries
calculate_revenue_split: bool = True # Split payment per agreement
# Overpayment
overpayment_action: str # "credit" | "refund" | "manual"
7.4 Integration with financial ecosystem¶
Payment import connects to these modules:
Payment file import
│
├── SettlementOrderService.apply_payment() → Allocate by avräkningsordning
├── AuditLogger.log_insert() → Audit trail
├── JournalEntry (bookkeeping) → Debit bank, credit receivable
├── RevenueSplitService.calculate_split() → Split per agreement
└── AccountingExportProvider.export() → Sync to Fortnox/Visma
8. Data Models¶
Import run tracking¶
class ImportRun(BaseDocument):
"""Record of a file import execution."""
tenant_id: PydanticObjectId
configuration_id: PydanticObjectId
configuration_name: str
# File info
filename: str
file_size_bytes: int
file_format: str
source_type: str # "sftp" | "upload" | "directory"
# Status
status: str # "processing" | "completed" | "completed_with_errors" | "failed"
started_at: datetime
completed_at: Optional[datetime] = None
# Results
total_records: int = 0
created: int = 0
updated: int = 0
deleted: int = 0
skipped: int = 0
failed: int = 0
unmatched: int = 0 # For payment files
# Errors
errors: list[ImportError] = []
class TenantConfig:
tenant_scoped = True
class Settings:
name = "import_runs"
class RejectedRecord(BaseDocument):
"""
A record that was rejected by the API during import.
Stored as a full document so the end user can:
1. See what went wrong (error from API)
2. See the original data (raw + mapped)
3. Edit the mapped data and retry
4. Delete the record (skip it)
Rejected records do NOT block the rest of the file.
"""
tenant_id: PydanticObjectId
import_run_id: PydanticObjectId
configuration_id: PydanticObjectId
# Position in file
record_number: int # Line/record number in the source file
post_type: Optional[str] = None # For flat files: prefix
# Data
raw_data: dict # As parsed from file (before mapping)
mapped_data: dict # After field mapping + transformation (what was sent to API)
# What was attempted
target_endpoint: str # "/customers"
http_method: str # "POST" | "PATCH" | "PUT" | "DELETE"
target_item_id: Optional[str] = None # For PATCH/PUT/DELETE
# Why it failed
error_code: int # HTTP status: 422, 400, 409, etc.
error_message: str # Human-readable error
error_detail: Optional[dict] = None # Full API error response (validation errors, etc.)
field_errors: list[FieldError] = [] # Per-field errors for form display
# Resolution
status: str = "rejected"
# "rejected" → Needs attention
# "corrected" → User edited mapped_data, ready to retry
# "retried" → Retry attempted
# "resolved" → Successfully imported after correction
# "skipped" → User chose to skip/delete this record
corrected_data: Optional[dict] = None # User's corrected version
resolved_at: Optional[datetime] = None
resolved_by: Optional[PydanticObjectId] = None
class TenantConfig:
tenant_scoped = True
class Settings:
name = "import_rejected_records"
indexes = [
[("import_run_id", 1), ("status", 1)],
[("tenant_id", 1), ("status", 1)],
]
class FieldError(BaseModel):
"""A validation error for a specific field."""
field: str # "email", "amount", "organization_id"
message: str # "Invalid email format", "Value must be positive"
value: Optional[Any] = None # The rejected value
class UnmatchedPayment(BaseDocument):
"""Payment from a file that couldn't be matched to a claim."""
tenant_id: PydanticObjectId
import_run_id: PydanticObjectId
reference: str
amount: Decimal
currency: str
date: datetime
raw_data: dict
status: str # "pending" | "manually_matched" | "refunded" | "written_off"
matched_claim_id: Optional[PydanticObjectId] = None
class TenantConfig:
tenant_scoped = True
class Settings:
name = "unmatched_payments"
9. Error Handling and Correction¶
9.1 Principle: never block the whole file¶
A file with 10 000 records where 3 have validation errors must still import the 9 997 valid records. The 3 rejected records are stored for review and correction.
9.2 Import flow with error handling¶
File with 500 records
│
▼
Parse all records
│
▼
For each record:
│
├─ Map fields + transform
├─ POST/PATCH to API
│
├─ 2xx Success → count as created/updated
│
└─ 4xx Error → store as RejectedRecord
│
├─ 422 Validation error → save field_errors
├─ 409 Conflict → save error_message
├─ 400 Bad request → save error_detail
└─ Continue with next record
│
▼
Import complete
Status: "completed_with_errors"
Result: 497 created, 3 rejected
│
▼
Notification → tenant admin
"Import completed: 497 of 500 records imported. 3 need attention."
9.3 Correction flow (end user in admin)¶
Admin UI: File Import → Import Runs → "Customer import 2026-03-28"
│
▼
Summary: 497 created, 3 rejected
│
Click "3 rejected"
│
▼
List of rejected records:
┌──────┬────────────────────┬──────────────────────────┬──────────┐
│ # │ Data │ Error │ Actions │
├──────┼────────────────────┼──────────────────────────┼──────────┤
│ 42 │ name: "AB Corp" │ email: Invalid format │ Edit │ ✕ │
│ │ email: "not-email" │ │ │
├──────┼────────────────────┼──────────────────────────┼──────────┤
│ 198 │ name: "" │ name: Required field │ Edit │ ✕ │
│ │ email: "a@b.com" │ │ │
├──────┼────────────────────┼──────────────────────────┼──────────┤
│ 331 │ name: "Dup AB" │ external_id: Already │ Edit │ ✕ │
│ │ ext_id: "C-100" │ exists (conflict) │ │
└──────┴────────────────────┴──────────────────────────┴──────────┘
User clicks "Edit" on record #42:
│
▼
Form pre-filled with mapped_data, field errors highlighted:
Name: [AB Corp ]
Email: [not-email ] ← "Invalid email format"
Phone: [+46701234567 ]
User corrects email → "info@abcorp.se" → clicks "Save & Retry"
│
▼
System: PATCH /import-rejected-records/{id} with corrected_data
→ status: "corrected"
│
▼
System: POST /customers with corrected_data
│
├─ Success → status: "resolved"
└─ Still fails → status: "rejected" with new error
User clicks "✕" (skip) on record #198:
→ status: "skipped"
9.4 API endpoints for rejected records¶
# List rejected records for an import run
GET /file-import/rejected?import_run_id={id}&status=rejected
# Get single rejected record (full detail)
GET /file-import/rejected/{id}
# Correct and retry
PATCH /file-import/rejected/{id}
{
"corrected_data": {"email": "info@abcorp.se"},
"action": "retry"
}
# Skip/delete rejected record
PATCH /file-import/rejected/{id}
{
"action": "skip"
}
# Retry all corrected records in one batch
POST /file-import/rejected/retry-all?import_run_id={id}
# Skip all remaining rejected records
POST /file-import/rejected/skip-all?import_run_id={id}
9.5 Retry logic¶
When retrying a corrected record:
async def retry_rejected_record(record: RejectedRecord) -> bool:
"""Retry a corrected rejected record against the API."""
data = record.corrected_data or record.mapped_data
try:
if record.http_method == "POST":
await api.post(record.target_endpoint, data)
elif record.http_method == "PATCH":
await api.patch(f"{record.target_endpoint}/{record.target_item_id}", data)
# ... PUT, DELETE
record.status = "resolved"
record.resolved_at = datetime.now(UTC)
await record.save()
# Update import run counts
run = await ImportRun.get(record.import_run_id)
run.failed -= 1
run.created += 1 # or updated, depending on method
await run.save()
return True
except APIError as e:
# Still failing — update error, keep as rejected
record.error_code = e.status_code
record.error_message = str(e)
record.error_detail = e.detail
record.status = "rejected"
record.corrected_data = data # Keep the correction
await record.save()
return False
9.6 Admin UI integration¶
The admin app (craft-easy-admin) shows:
| View | What | Where |
|---|---|---|
| Import runs list | All imports with status badge (green/yellow/red) | /admin/file-import |
| Import run detail | Summary + record counts + rejected list | /admin/file-import/{run_id} |
| Rejected record form | Pre-filled form with field errors highlighted | /admin/file-import/{run_id}/rejected/{id} |
| Bulk actions | "Retry all corrected" / "Skip all remaining" | /admin/file-import/{run_id} |
| Notification badge | "3 rejected records need attention" | Sidebar/header |
10. Flows and Examples¶
10.1 Nightly customer sync from ERP¶
1. ERP exports customers.json to SFTP every night
2. File import polls SFTP at 02:00
3. Detects new file → matches import config "ERP Customer Sync"
4. Parses JSON → 500 customer records
5. Sync mode: "upsert" with sync_key: "external_id"
6. Result: 12 created, 488 updated, 0 errors
7. Post-import job: bi-export (refresh BI data)
10.2 Bankgirot payment file¶
1. Bank uploads BGC file to SFTP at 15:00
2. File import detects PAYMENT_20260328.dat
3. Parses positional format → 45 payments
4. For each payment:
- Match OCR to claim.payment_reference
- Apply via settlement order
- Create journal entry
- Calculate revenue split
5. Result: 42 matched, 3 unmatched
6. Unmatched → notification to tenant admin in admin app
7. Admin manually matches or marks for refund
10.3 XML product catalog update¶
1. Tenant uploads catalog.xml via admin UI
2. File import parses XML with XPath config
3. Sync mode: "sync" — creates, updates, AND deletes
4. 200 products in file, 210 in system
5. Result: 5 created, 195 updated, 10 deleted
11. Implementation Plan¶
| Phase | What | Weeks | Depends on |
|---|---|---|---|
| 1 | Core: parsers (JSON, CSV, positional, XML) | 2 | craft-easy-api |
| 2 | Import config model + field mapping + sync engine | 2 | Phase 1 |
| 3 | SFTP connector + polling (via craft-easy-jobs) | 1 | Phase 2 + craft-easy-jobs |
| 4 | File upload API endpoint | 0.5 | Phase 2 |
| 5 | Payment file import + reconciliation | 2 | Phase 2 + financial modules |
| 6 | Admin UI integration (import config, run history, unmatched) | 1 | Phase 2 + craft-easy-admin |
Total: ~8-9 weeks
Ecosystem (updated)¶
PyPI packages:
┌──────────────────────────────────────┐
│ craft-easy-api (foundation) │
└──────────────┬───────────────────────┘
│ depends on
┌──────────┼──────────────────────────┐
│ │ │
┌───▼────────┐ ┌▼────────────────┐ ┌─────▼──────────────┐
│ craft-easy │ │ craft-easy │ │ craft-easy │
│ -jobs │ │ -file-import │ │ -admin │
│ Batch │ │ SFTP/XML/JSON │ │ Universal admin UI │
│ processing │ │ → API endpoints │ │ (React/Expo) │
└────────────┘ └─────────────────┘ └─────────────────────┘
GitHub repos (easy-software-system/):
├── craft-easy-api ✅ Created + scaffolded + code
├── craft-easy-admin ✅ Created (empty)
├── craft-easy-jobs ✅ Created (empty)
├── craft-easy-file-import ✅ Created (empty)
└── craft-easy-template ✅ Created (empty)
PyPI:
├── craft-easy-api v0.1.0 ✅ Published
├── craft-easy-jobs v0.0.1 ✅ Reserved
├── craft-easy-file-import v0.0.1 ✅ Reserved
└── craft-easy-admin v0.0.1 ✅ Reserved (npm)