API Reference

API Reference#

Props#

class port.api.props.PropsUIDataSubmissionButtons(donate_question: Translatable | None = None, donate_button: Translatable | None = None, waiting: bool = False)#

Buttons for data submission actions

donate_question#

Optional question text above buttons

Type:: port.api.props.Translatable | None

donate_button#

Optional text for donate button

Type:: port.api.props.Translatable | None

waiting#

Whether the data submission is in progress

Type:: bool

class port.api.props.PropsUIFooter(progressPercentage: float)#

Page footer

progressPercentage#

float indicating the progress in the flow

Type:: float

class port.api.props.PropsUIHeader(title: Translatable)#

Page header

title#

title of the page

Type:: port.api.props.Translatable

A multi-purpose page that gets shown to the user

platform#

the platform name the user is curently in the process of donating data from

Type:: str

header#

page header

Type:: port.api.props.PropsUIHeader

body#

main body of the page, see the individual classes for an explanation

Type:: port.api.props.PropsUIPromptRadioInput | port.api.props.PropsUIPromptConsentForm | port.api.props.PropsUIPromptFileInput | port.api.props.PropsUIPromptConfirm | port.api.props.PropsUIPromptProgress | port.api.props.PropsUIPromptHelloWorld | port.api.props.PropsUIPromptConsentFormTable | port.api.props.PropsUIDataSubmissionButtons | list | Any

class port.api.props.PropsUIPageError(message: str)#: An error page to show when something goes wrong

class port.api.props.PropsUIPromptConfirm(text: Translatable, ok: Translatable, cancel: Translatable | None = None)#

Retry submitting a file page

Prompt the user if they want to submit a new file. This can be used in case a file could not be processed.

text#

message to display

Type:: port.api.props.Translatable

ok#

message to display if the user wants to try again

Type:: port.api.props.Translatable

cancel#

message to display if the user wants to continue regardless

Type:: port.api.props.Translatable | None

class port.api.props.PropsUIPromptConsentForm(tables: list[PropsUIPromptConsentFormTable], description: Translatable | None = None, donate_question: Translatable | None = None, donate_button: Translatable | None = None)#

Tables to be shown to the participant prior to data submission

tables#

a list of tables, including both editable and read-only tables

Type:: list[port.api.props.PropsUIPromptConsentFormTable]

description#

Optional description text

Type:: port.api.props.Translatable | None

donate_question#

Optional question text for data submission button

Type:: port.api.props.Translatable | None

donate_button#

Optional text for data submission button

Type:: port.api.props.Translatable | None

class port.api.props.PropsUIPromptConsentFormTable(id: str, number: int, title: Translatable, description: Translatable, data_frame: pandas.DataFrame, data_frame_max_size: int = 10000, headers: dict[str, Translatable] | None = None)#

Table to be shown to the participant prior to data_submission

It is truncated to a maximum number of rows to avoid overloading the UI.

id#

a unique string to itentify the table after donation

Type:: str

number#

the number of table in the list of tables

Type:: int

title#

title of the table

Type:: port.api.props.Translatable

description#

description of the table

Type:: port.api.props.Translatable

data_frame#

table to be shown

Type:: pandas.DataFrame

data_frame_max_size#

maximum size of the table (in rows)

Type:: int

headers#

optional headers for the table columns

Type:: dict[str, port.api.props.Translatable] | None

class port.api.props.PropsUIPromptFileInput(description: Translatable, extensions: str)#

Prompt the user to submit a file

description#

text with an explanation

Type:: port.api.props.Translatable

extensions#

accepted mime types, example: “application/zip, text/plain”

Type:: str

class port.api.props.PropsUIPromptHelloWorld(text: Translatable)#

Hello world component to welcome users

text#

welcome message to display

Type:: port.api.props.Translatable

class port.api.props.PropsUIPromptProgress(description: Translatable, message: str, percentage: int | None = None)#

Prompt the user information during the extraction

description#

text with an explanation

Type:: port.api.props.Translatable

message#

can be used to show extraction progress

Type:: str

class port.api.props.PropsUIPromptRadioInput(title: Translatable, description: Translatable, items: list[RadioItem])#

Radio group

This radio group can be used get a mutiple choice answer from a user

title#

title of the radio group

Type:: port.api.props.Translatable

description#

short description of the radio group

Type:: port.api.props.Translatable

items#

a list of radio buttons

Type:: list[port.api.props.RadioItem]

class port.api.props.PropsUIPromptText(text: Translatable, title: Translatable | None = None)#

Text block to display information to the user

title#

optional title for the text block

Type:: port.api.props.Translatable | None

text#

main text content to display

Type:: port.api.props.Translatable

class port.api.props.RadioItem#

Radio button

id#

id of radio button

Type:: int

value#

text to be displayed

Type:: str

class port.api.props.Translatable(translations: Translations)#: Wrapper class for Translations

class port.api.props.Translations#

Typed dict containing text displayed in a specific language.

en and nl are required. Additional languages are optional. The feldspar Translator falls back gracefully for missing locales.

en#

English string to display

Type:: str

nl#

Dutch string to display

Type:: str

es#

Spanish string to display (optional)

Type:: str

Extraction helpers#

This module contains helper functions that can be used during the data extraction process

class port.helpers.extraction_helpers.CsvExtractionResult(found: bool, data: pandas.DataFrame, member_path: str | None = None)#: Result of extracting and parsing a CSV file from a zip.

exception port.helpers.extraction_helpers.FileNotFoundInZipError#: The File you are looking for is not present in a zipfile

class port.helpers.extraction_helpers.JsonExtractionResult(found: bool, data: dict | list, member_path: str | None = None)#: Result of extracting and parsing a JSON file from a zip.

class port.helpers.extraction_helpers.RawExtractionResult(found: bool, data: BytesIO, member_path: str | None = None)#: Result of extracting raw bytes from a zip.

class port.helpers.extraction_helpers.ZipArchiveReader(archive: SeekableBinaryReader, archive_members: list[str], errors: Counter)#

Reads files from a zip archive using cached member inventory.

Encapsulates a seekable binary archive (SeekableBinaryReader), archive member list (from validation), and error counter. Provides json()/csv()/raw() methods with found/not-found signaling to eliminate cascading errors for expected-missing files.

Per ADR-0026, the upload pipeline passes the AsyncFileAdapter from a browser upload here directly so the zip is never materialized into Pyodide’s heap. Path-string inputs are not accepted; tests construct fixtures via io.BytesIO.

Usage:: reader = ZipArchiveReader(archive, validation.archive_members, errors) result = reader.json(“following.json”) if result.found:

data = result.data # parsed dict/list

csv(filename: str) → CsvExtractionResult#

Extract and parse a CSV file.

Returns CsvExtractionResult(found=False, data=pd.DataFrame()) if member not in archive.

json(filename: str) → JsonExtractionResult#

Extract and parse a JSON file.

Returns JsonExtractionResult(found=False, data={}) if member not in archive. Skips JSON parsing entirely when not found.

json_all(pattern: str) → list[JsonExtractionResult]#

Extract and parse all JSON files matching a regex pattern.

Returns results sorted lexicographically by member path. Used for paginated exports (post_comments_1.json, _2.json, etc.).

raw(filename: str) → RawExtractionResult#

Extract raw bytes from a zip member.

Returns RawExtractionResult(found=False, data=io.BytesIO()) if member not in archive. Used for HTML (Chrome bookmarks), text files (WhatsApp, TikTok), and .js files (X — caller applies bytesio_to_listdict for JS prefix stripping).

resolve_member(filename: str) → str | None#

Resolve a filename to an archive member path.

Resolution rule: 1. Exact path match → use it. 2. Path-boundary suffix match (member.endswith(“/” + filename)) →

if exactly 1, use it.

0 matches → return None.
Multiple matches → return None, log warning, increment errors[“AmbiguousMemberMatch”].

port.helpers.extraction_helpers.dict_denester(inp: dict[Any, Any] | list[Any], new: dict[Any, Any] | None = None, name: str = '', run_first: bool = True) → dict[Any, Any]#

Denests a dictionary or list, returning a new flattened dictionary.

Parameters:

inp (dict[Any, Any] | list[Any]) – The input dictionary or list to be denested.
new (dict[Any, Any] | None, optional) – The dictionary to store denested key-value pairs. Defaults to None.
name (str, optional) – The current key name in the denesting process. Defaults to “”.
run_first (bool, optional) – Flag to indicate if this is the first run of the function. Defaults to True.

Returns:

A new denested dictionary.

Return type:

dict[Any, Any]

Examples:

>>> nested_dict = {"a": {"b": {"c": 1}}, "d": [2, 3]}
>>> dict_denester(nested_dict)
{"a-b-c": 1, "d-0": 2, "d-1": 3}

port.helpers.extraction_helpers.epoch_to_iso(epoch_timestamp: str | int | float, errors: Counter | None = None) → str#

Convert epoch timestamp to an ISO 8601 string, assuming UTC.

Parameters:: epoch_timestamp (str | int) – The epoch timestamp to convert.
Returns:: The ISO 8601 formatted string, or the original input if conversion fails.
Return type:: str
Raises:: Exception – Logs an error message if conversion fails.

Examples:

>>> epoch_to_iso(1632139200)
"2021-09-20T12:00:00+00:00"

port.helpers.extraction_helpers.extract_file_from_zip(zfile: str, file_to_extract: str, errors: Counter | None = None) → BytesIO#

Extracts a specific file from a zipfile and returns it as a BytesIO buffer.

Parameters:

zfile (str) – Path to the zip file.
file_to_extract (str) – Name or path of the file to extract from the zip.
errors (Counter | None) – Optional counter for aggregating error types.

Returns:

A BytesIO buffer containing the extracted file’s content of the first file found.: Returns an empty BytesIO if the file is not found or an error occurs.

Return type:

io.BytesIO

port.helpers.extraction_helpers.find_item(d: dict[Any, Any], key_to_match: str) → str#

Finds the least nested value in a denested dictionary whose key contains the given key_to_match.

Parameters:

d (dict[Any, Any]) – A denested dictionary to search in.
key_to_match (str) – The substring to match in the keys.

Returns:

The value of the least nested key containing key_to_match.: Returns an empty string if no match is found.

Return type:

str

Raises:

Exception – Logs an error message if an exception occurs during the search.

Examples:

>>> d = {"asd-asd-asd": 1, "asd-asd": 2, "qwe": 3}
>>> find_item(d, "asd")
"2"

port.helpers.extraction_helpers.find_items(d: dict[Any, Any], key_to_match: str) → list#

Finds all values in a denested dictionary whose keys contain the given key_to_match.

Parameters:

d (dict[Any, Any]) – A denested dictionary to search in.
key_to_match (str) – The substring to match in the keys.

Returns:

A list of all values whose keys contain key_to_match.

Return type:

list

Raises:

Exception – Logs an error message if an exception occurs during the search.

Examples:

>>> d = {"asd-1": "a", "asd-2": "b", "qwe": "c"}
>>> find_items(d, "asd")
["a", "b"]

port.helpers.extraction_helpers.fix_ascii_string(input: str) → str#

Fixes the string encoding by removing non-ASCII characters.

Parameters:: input (str) – The input string that needs to be fixed.
Returns:: The fixed string with only ASCII characters, or the original string if an exception occurs.
Return type:: str

Examples:

>>> fix_ascii_string("Hello, 世界!")
"Hello, !"

port.helpers.extraction_helpers.fix_latin1_string(input: str) → str#

Fixes the string encoding by attempting to encode it using the ‘latin1’ encoding and then decoding it.

Parameters:: input (str) – The input string that needs to be fixed.
Returns:: The fixed string after encoding and decoding, or the original string if an exception occurs.
Return type:: str

Examples:

>>> fix_latin1_string("café")
"café"

port.helpers.extraction_helpers.json_dumper(zfile: str) → pandas.DataFrame#

Reads all JSON files in a zip file, flattens them, and combines them into a single DataFrame.

Parameters:: zfile (str) – Path to the zip file containing JSON files.
Returns:: A DataFrame containing flattened data from all JSON files in the zip.
Return type:: pd.DataFrame
Raises:: Exception – Logs an error message if an exception occurs during the process.

Examples:

>>> df = json_dumper("data.zip")
>>> print(df.head())

port.helpers.extraction_helpers.read_csv_from_bytes(json_bytes: BytesIO, errors: Counter | None = None) → list[dict[Any, Any]]#

Reads CSV data from a BytesIO buffer and returns it as a list of dictionaries.

Parameters:

json_bytes (io.BytesIO) – A BytesIO buffer containing CSV data.
errors (Counter | None) – Optional counter for aggregating error types.

Returns:

A list of dictionaries, where each dictionary represents a row in the CSV.: Returns an empty list if parsing fails.

Return type:

list[dict[Any, Any]]

port.helpers.extraction_helpers.read_csv_from_bytes_to_df(json_bytes: BytesIO) → pandas.DataFrame#

Reads CSV data from a BytesIO buffer and returns it as a pandas DataFrame.

Parameters:: json_bytes (io.BytesIO) – A BytesIO buffer containing CSV data.
Returns:: A pandas DataFrame containing the CSV data.
Return type:: pd.DataFrame

Examples

>>> buffer = io.BytesIO(b'name,age\nAlice,30\nBob,25')
>>> df = read_csv_from_bytes_to_df(buffer)
>>> print(df)
   name  age
0  Alice   30
1    Bob   25

port.helpers.extraction_helpers.read_json_from_bytes(json_bytes: BytesIO, errors: Counter | None = None) → dict[Any, Any] | list[Any]#

Reads JSON data from a BytesIO buffer.

Parameters:

json_bytes (io.BytesIO) – A BytesIO buffer containing JSON data.

Returns:

The parsed JSON data as a dictionary or list.: Returns an empty dictionary if parsing fails.

Return type:

dict[Any, Any] | list[Any]

Examples:

>>> buffer = io.BytesIO(b'{"key": "value"}')
>>> data = read_json_from_bytes(buffer)
>>> print(data)
{'key': 'value'}

port.helpers.extraction_helpers.read_json_from_file(json_file: str) → dict[Any, Any] | list[Any]#

Reads JSON data from a file.

Parameters:

json_file (str) – Path to the JSON file.

Returns:

The parsed JSON data as a dictionary or list.: Returns an empty dictionary if parsing fails.

Return type:

dict[Any, Any] | list[Any]

Examples:

>>> data = read_json_from_file("data.json")
>>> print(data)
{'key': 'value'}

port.helpers.extraction_helpers.replace_months(input_string: str) → str#

Replaces Dutch month abbreviations with English equivalents in the input string.

Parameters:: input_string (str) – The input string containing potential Dutch month abbreviations.
Returns:: The input string with Dutch month abbreviations replaced by English equivalents.
Return type:: str

Examples:

>>> replace_months("15 mei 2023")
"15 may 2023"

port.helpers.extraction_helpers.sort_isotimestamp_empty_timestamp_last(timestamp_series: pandas.Series) → pandas.Series#

Creates a key for sorting a pandas Series of ISO timestamps, placing empty timestamps last.

Parameters:: timestamp_series (pd.Series) – A pandas Series containing ISO formatted timestamps.
Returns:: A Series of sorting keys, with -timestamp for valid dates and infinity for invalid/empty dates.
Return type:: pd.Series

Examples:

>>> df = df.sort_values(by="Date", key=sort_isotimestamp_empty_timestamp_last)

Port helpers#

port.helpers.port_helpers.donate(key: str, json_string: str) → CommandSystemDonate#

Initiates a donation process using the provided key and data.

This function triggers the donation process by passing a key and a JSON-formatted string that contains donation information.

Parameters:

key (str) – The key associated with the donation process. The key will be used in the file name.
json_string (str) – A JSON-formatted string containing the donated data.

Returns:

A system command that initiates the donation process. Must be yielded.

Return type:

CommandSystemDonate

port.helpers.port_helpers.emit_log(level: str, message: str)#

Yield a CommandSystemLog to the host via the command protocol.

Use via yield from emit_log(…) in generators (FlowBuilder, script.py). The host receives the log immediately; the PayloadVoid response is discarded.

Messages sent through this function reach mono’s /api/feldspar/log. They MUST be PII-free — no file paths, exception text, or participant data.

Examples:

yield from emit_log("info", "[LinkedIn] Consent: accepted")
yield from emit_log("info", "Starting platform: Facebook")

port.helpers.port_helpers.exit(code: int, info: str) → CommandSystemExit#

Exits Next with the provided exit code and additional information. This if the code reaches this function, it will return to the task list in Next.

Parameters:

code (int) – The exit code representing the type or status of the exit.
info (str) – A string containing additional information about the exit.

Returns:

A system command that initiates the exit process in Next.

Return type:

CommandSystemExit

Examples:

yield exit(0, "Success")

port.helpers.port_helpers.generate_file_prompt(extensions: str, multiple: bool = False) → PropsUIPromptFileInput | PropsUIPromptFileInputMultiple#

Generates a file input prompt for selecting file(s) for a platform. This function creates a bilingual (English and Dutch) file input prompt that instructs the user to select file(s) they’ve received from a platform and stored on their device.

The prompt that is returned by this function needs to be rendered using: yield result = render_page(…) result.value should then contain the file handle(s). In case multiple is true, a list with file handles is returned.

Parameters:

extensions (str) – A collection of allowed MIME types. For example: “application/zip, text/plain, application/json”
multiple (bool, optional) – Whether to allow multiple file selection. Defaults to False.

Returns:

A file input prompt object containing the description text and allowed file extensions. If multiple=True, returns a PropsUIPromptFileInputMultiple object for selecting multiple files.

Return type:

props.PropsUIPromptFileInput | d3i_props.PropsUIPromptFileInputMultiple

port.helpers.port_helpers.generate_questionnaire() → PropsUIPromptQuestionnaire#

Administer a basic questionnaire in Port.

This function generates a prompt which can be rendered with render_page(). The questionnaire demonstrates all currently implemented question types. In the current implementation, all questions are optional.

You can build in logic by: - Chaining questionnaires together - Using extracted data in your questionnaires

Usage:: prompt = generate_questionnaire() results = yield render_page(header_text, prompt)

The results.value contains a JSON string with question answers that can then be donated with donate().

port.helpers.port_helpers.generate_radio_prompt(title: Translatable, description: Translatable, items: list[str]) → PropsUIPromptRadioInput#: General purpose prompt selection menu

port.helpers.port_helpers.generate_retry_prompt(platform_name: str) → PropsUIPromptConfirm#

Generate a bilingual retry prompt for file processing errors.

Returns a PropsUIPromptConfirm with “Try again” (ok → PayloadTrue) and “Continue” (cancel → PayloadFalse) buttons. Using standard feldspar PropsUIPromptConfirm instead of d3i PropsUIPromptRetry which only renders a single button. See ADR-0016 for the broader decision on custom vs standard prompt components.

Parameters:: platform_name – The name of the platform whose file could not be processed.

port.helpers.port_helpers.generate_review_data_prompt(description: Translatable, table_list: list[PropsUIPromptConsentFormTableViz]) → PropsUIPromptConsentFormViz#

Generates a data review form with a list of tables and a description, including default donate question and button. The participant can review these tables before they will be send to the researcher. If the participant consents to sharing the data the data will be stored at the configured storage location.

Parameters:

table_list (list[props.PropsUIPromptConsentFormTableViz]) – A list of consent form tables to be included in the prompt.
description (props.Translatable) – A translatable description text for the consent prompt.

Returns:

A structured consent form object containing the provided table list, description, and default values for donate question and button.

Return type:

props.PropsUIPromptConsentForm

port.helpers.port_helpers.handle_donate_result(result) → bool#

Inspect donate result. Returns True on success, False on failure.

eyra/feldspar develop (Feb 2026+) returns PayloadResponse for CommandSystemDonate with value.success indicating outcome. Older feldspar and FakeBridge (dev mode) return PayloadVoid (fire-and-forget).

PayloadResponse → check value.success (production path, checked first) PayloadVoid / None → True (dev mode / backward-compat) Anything else → log warning, return False

port.helpers.port_helpers.render_donate_failure_page(platform_name: str) → CommandUIRender#

Render donation failure page.

Caller should yield and await response before returning.

port.helpers.port_helpers.render_no_data_page(platform_name: str) → CommandUIRender#

Render ‘no relevant data found’ with acknowledge button.

Caller should yield and await response before returning.

Renders the UI components for a donation page.

This function assembles various UI components including a header, body, and footer to create a complete donation page. It uses the provided header text and body content to customize the page.

Parameters:

header_text (props.Translatable) – The text to be displayed in the header. This should be a translatable object to support multiple languages.
( (body) – props.PropsUIPromptRadioInput | props.PropsUIPromptConsentForm | props.PropsUIPromptFileInput | props.PropsUIPromptConfirm |
) – The main content of the page. It must be compatible with props.PropsUIPageDonation.

Returns:

A render command object containing the fully assembled page. Must be yielded.

Return type:

CommandUIRender

port.helpers.port_helpers.render_safety_error_page(platform_name: str, error: Exception) → CommandUIRender#

Render file safety error page.

Caller should yield and await response before returning.

Validation#

Contains classes to deal with input validation of DDPs

The idea of this module is to provide a uniform way to assign a validation status to a DDP validation Which can be used and acted upon

class port.helpers.validate.BaseValidation(status_code: int)#: Base validation class that can be used for validation purposes

class port.helpers.validate.DDPCategory(id: str, ddp_filetype: DDPFiletype, language: Language, known_files: list[str])#

Represents characteristics that define a DDP (Data Delivery Package) category.

Parameters:

id (str) – Unique identifier for the DDP category.
ddp_filetype (DDPFiletype) – The file type of the DDP.
language (Language) – The language of the DDP.
known_files (List[str]) – A list of known files associated with this DDP category.

Examples

>>> category = DDPCategory("cat1", DDPFiletype.JSON, Language.EN, ["file1.json", "file2.json"])
>>> print(category.id)
cat1
>>> print(category.language)
<Language.EN: 1>

class port.helpers.validate.DDPFiletype(*values)#: Enumeration of supported DDP file types.

class port.helpers.validate.Language(*values)#: Enumeration of supported languages.

class port.helpers.validate.StatusCode(id: int, description: str)#

Represents a status code that can be used to set a DDP status.

Parameters:

id (int) – The numeric identifier of the status code.
description (str) – A brief description of what the status code represents.

Examples

>>> status = StatusCode(0, "Success")
>>> print(status.id)
0
>>> print(status.description)
Success

class port.helpers.validate.ValidateInput(all_status_codes: list[~port.helpers.validate.StatusCode], all_ddp_categories: list[~port.helpers.validate.DDPCategory], current_status_code: ~port.helpers.validate.StatusCode | None = None, current_ddp_category: ~port.helpers.validate.DDPCategory | None = None, archive_members: list[str] = <factory>)#

A class for validating input data against predefined categories and status codes.

Parameters:

all_status_codes (List[StatusCode]) – A list of valid status codes.
all_ddp_categories (List[DDPCategory]) – A list of valid DDP categories.
current_status_code (Optional[StatusCode]) – The current status code. Defaults to None.
current_ddp_category (Optional[DDPCategory]) – The current DDP category. Defaults to None.

ddp_categories_lookup#

A lookup dictionary for DDP categories.

Type:: Dict[str, DDPCategory]

status_codes_lookup#

A lookup dictionary for status codes.

Type:: Dict[int, StatusCode]

Examples

>>> status_codes = [StatusCode(id=0, description="Success"), StatusCode(id=1, description="Error")]
>>> ddp_categories = [DDPCategory(id="cat1", ddp_filetype=DDPFiletype.JSON, language=Language.EN, known_files=["file1.txt", "file2.txt"])]
>>> validator = ValidateInput(all_status_codes=status_codes, all_ddp_categories=ddp_categories)

get_status_code_id() → int#

Return the current assigned status code ID. Note: zero is always used for OK. Non-zero otherwise.

Returns:: The ID of the current status code, or 1 if no status code is set.
Return type:: int

Examples

>>> validator.get_status_code_id()

infer_ddp_category(file_list_input: list[str]) → bool#

Compares a list of files to a list of known files and infers the DDPCategory.

Parameters:: file_list_input (List[str]) – A list of input files to compare against known files.
Returns:: True if a valid DDP category is inferred, False otherwise. It sets the current_status_code and current_ddp_category to either the DDP catogory match, or to an unknown category.
Return type:: bool

Examples

>>> validator.infer_ddp_category(["file1.txt", "file2.txt"])

set_current_status_code_by_id(id: int) → None#

Set the status code based on the provided ID.

Parameters:: id (int) – The ID of the status code to set.

Examples

>>> validator.set_current_status_code_by_id(0)

port.helpers.validate.validate_zip(ddp_categories: list[DDPCategory], archive: SeekableBinaryReader) → ValidateInput#

Validates a DDP zip archive against a list of DDP categories.

This function attempts to open and read the contents of a zip archive, then uses the ValidateInput class to infer the DDP category based on the files in the zip. If the archive is invalid or cannot be read, it sets an error status code (an integer greater than 0).

Parameters:

ddp_categories (List[DDPCategory]) – A list of valid DDP categories to compare against.
archive – A seekable binary file-like object — typically an AsyncFileAdapter from a browser upload, or an io.BytesIO in tests. Per ADR-0026, the upload pipeline never materializes uploads to a path; consumers accept the file-like adapter directly to avoid FileReaderSync’s ~2 GiB cap.

Returns:

An instance of ValidateInput containing the: validation results.

Return type:

ValidateInput

Raises:

zipfile.BadZipFile – This exception is caught internally and results in an error status code.

FlowBuilder#

FlowBuilder — shared per-platform donation flow orchestration.

Subclass this to implement a platform-specific donation flow. Override validate_file() and extract_data(). Call start_flow() as a generator from script.py via yield from.

Upload safety#

Upload safety checks.

Validates upload size against policy limits using metadata only — the upload itself is never read into Pyodide’s heap.

See ADR-0026 for the streaming invariant: PayloadFile uploads must be passed directly to consumers (zipfile.ZipFile, validators, extractors) without materialization. Reading the entire payload to verify its size defeats this; the JS-reported adapter.size attribute is the source of truth for size policy decisions.

exception port.helpers.uploads.ChunkedExportError#: Raised when a file is exactly CHUNKED_EXPORT_SENTINEL_BYTES (split export sentinel).

exception port.helpers.uploads.FileTooLargeError#: Raised when a file exceeds MAX_FILE_SIZE_BYTES.

port.helpers.uploads.check_payload_size(file_result) → None#

Validate upload size from JS-reported metadata. No bytes read.

Caller is expected to handle the exception and render a safety error page. FlowBuilder does this around step 1 of start_flow().

Parameters:

file_result – A PayloadFile-shaped object whose .value carries an AsyncFileAdapter (with a .size attribute populated from the JS reader at construction time).

Raises:

TypeError – If file_result is not a PayloadFile. PayloadString / WORKERFS support was retired with ADR-0026.
ChunkedExportError – If size == CHUNKED_EXPORT_SENTINEL_BYTES (split export sentinel — incomplete multi-part download).
FileTooLargeError – If size > MAX_FILE_SIZE_BYTES.

API Reference

Contents

API Reference#

Props#

Extraction helpers#

Port helpers#

Validation#

FlowBuilder#

Upload safety#