# This is the main code for the `./manage.py export` data export tool.
# User docs: https://zulip.readthedocs.io/en/latest/production/export-and-import.html
#
# Most developers will interact with this primarily when they add a
# new table to the schema, in which case they likely need to (1) add
# it the lists in `ALL_ZULIP_TABLES` and similar data structures and
# (2) if it doesn't belong in EXCLUDED_TABLES, add a Config object for
# it to get_realm_config.
import glob
import hashlib
import logging
import os
import random
import secrets
import shutil
import subprocess
import tempfile
from collections.abc import Callable, Iterable, Iterator, Mapping
from contextvars import ContextVar
from dataclasses import dataclass
from datetime import datetime
from email.headerregistry import Address
from itertools import chain, islice
from typing import TYPE_CHECKING, Any, Optional, TypeAlias, TypedDict, TypeVar, cast
from urllib.parse import urlsplit

import orjson
from django.apps import apps
from django.conf import settings
from django.db import connection
from django.db.models import Exists, Model, OuterRef, Q, QuerySet
from django.forms.models import model_to_dict
from django.utils.timezone import is_naive as timezone_is_naive
from django.utils.timezone import now as timezone_now
from psycopg2 import sql

import zerver.lib.upload
from analytics.models import RealmCount, StreamCount, UserCount
from version import ZULIP_VERSION
from zerver.lib.avatar_hash import user_avatar_base_path_from_ids
from zerver.lib.display_recipient import get_display_recipient
from zerver.lib.migration_status import MigrationStatusJson, parse_migration_status
from zerver.lib.parallel import run_parallel, run_parallel_queue
from zerver.lib.pysa import mark_sanitized
from zerver.lib.stream_color import STREAM_ASSIGNMENT_COLORS
from zerver.lib.timestamp import datetime_to_timestamp
from zerver.lib.upload.s3 import get_bucket
from zerver.lib.utils import get_fk_field_name
from zerver.models import (
    AlertWord,
    Attachment,
    BotConfigData,
    BotStorageData,
    ChannelFolder,
    Client,
    CustomProfileField,
    CustomProfileFieldValue,
    DefaultStream,
    DirectMessageGroup,
    GroupGroupMembership,
    Message,
    MutedUser,
    NamedUserGroup,
    NavigationView,
    OnboardingStep,
    OnboardingUserMessage,
    Reaction,
    Realm,
    RealmAuditLog,
    RealmAuthenticationMethod,
    RealmDomain,
    RealmEmoji,
    RealmExport,
    RealmFilter,
    RealmPlayground,
    RealmUserDefault,
    Recipient,
    ScheduledMessage,
    Service,
    Stream,
    Subscription,
    UserActivity,
    UserActivityInterval,
    UserGroup,
    UserGroupMembership,
    UserMessage,
    UserPresence,
    UserProfile,
    UserStatus,
    UserTopic,
)
from zerver.models.messages import SubMessage
from zerver.models.presence import PresenceSequence
from zerver.models.realm_audit_logs import AuditLogEventType
from zerver.models.realms import get_fake_email_domain, get_realm
from zerver.models.saved_snippets import SavedSnippet
from zerver.models.users import ExternalAuthID, get_system_bot, is_cross_realm_bot_email

if TYPE_CHECKING:
    from mypy_boto3_s3.service_resource import Bucket, Object

# Custom mypy types follow:
Record: TypeAlias = dict[str, Any]
TableName = str
TableData: TypeAlias = dict[TableName, Iterator[Record] | list[Record]]
Field = str
Path = str
Context: TypeAlias = dict[str, Any]
FilterArgs: TypeAlias = dict[str, Any]
IdSource: TypeAlias = tuple[TableName, Field]
SourceFilter: TypeAlias = Callable[[Record], bool]

CustomFetch: TypeAlias = Callable[[TableData, Context], None]
CustomReturnIds: TypeAlias = Callable[[TableData], set[int]]
CustomProcessResults: TypeAlias = Callable[[Iterable[Record], Context], Iterator[Record]]


class MessagePartial(TypedDict):
    zerver_message: Iterable[Record]
    zerver_userprofile_ids: list[int]
    realm_id: int


ORJSON_ITERABLE_BATCH_SIZE = 1000
MESSAGE_BATCH_CHUNK_SIZE = 1000

ALL_ZULIP_TABLES = {
    "analytics_fillstate",
    "analytics_installationcount",
    "analytics_realmcount",
    "analytics_streamcount",
    "analytics_usercount",
    "otp_static_staticdevice",
    "otp_static_statictoken",
    "otp_totp_totpdevice",
    "social_auth_association",
    "social_auth_code",
    "social_auth_nonce",
    "social_auth_partial",
    "social_auth_usersocialauth",
    "two_factor_phonedevice",
    "zerver_alertword",
    "zerver_archivedattachment",
    "zerver_archivedattachment_messages",
    "zerver_archivedmessage",
    "zerver_archivedusermessage",
    "zerver_attachment",
    "zerver_attachment_messages",
    "zerver_attachment_scheduled_messages",
    "zerver_archivedreaction",
    "zerver_archivedsubmessage",
    "zerver_archivetransaction",
    "zerver_botconfigdata",
    "zerver_botstoragedata",
    "zerver_channelemailaddress",
    "zerver_channelfolder",
    "zerver_client",
    "zerver_customprofilefield",
    "zerver_customprofilefieldvalue",
    "zerver_defaultstream",
    "zerver_defaultstreamgroup",
    "zerver_defaultstreamgroup_streams",
    "zerver_draft",
    "zerver_emailchangestatus",
    "zerver_externalauthid",
    "zerver_groupgroupmembership",
    "zerver_huddle",
    "zerver_imageattachment",
    "zerver_message",
    "zerver_missedmessageemailaddress",
    "zerver_multiuseinvite",
    "zerver_multiuseinvite_streams",
    "zerver_multiuseinvite_groups",
    "zerver_namedusergroup",
    "zerver_navigationview",
    "zerver_onboardingstep",
    "zerver_onboardingusermessage",
    "zerver_preregistrationrealm",
    "zerver_preregistrationuser",
    "zerver_preregistrationuser_streams",
    "zerver_preregistrationuser_groups",
    "zerver_presencesequence",
    "zerver_pushdevice",
    "zerver_pushdevicetoken",
    "zerver_reaction",
    "zerver_realm",
    "zerver_realmauditlog",
    "zerver_realmauthenticationmethod",
    "zerver_realmcreationstatus",
    "zerver_realmdomain",
    "zerver_realmemoji",
    "zerver_realmexport",
    "zerver_realmfilter",
    "zerver_realmplayground",
    "zerver_realmreactivationstatus",
    "zerver_realmuserdefault",
    "zerver_recipient",
    "zerver_savedsnippet",
    "zerver_scheduledemail",
    "zerver_scheduledemail_users",
    "zerver_scheduledmessage",
    "zerver_scheduledmessagenotificationemail",
    "zerver_service",
    "zerver_stream",
    "zerver_submessage",
    "zerver_subscription",
    "zerver_useractivity",
    "zerver_useractivityinterval",
    "zerver_usergroup",
    "zerver_usergroupmembership",
    "zerver_usermessage",
    "zerver_userpresence",
    "zerver_userprofile",
    "zerver_userprofile_groups",
    "zerver_userprofile_user_permissions",
    "zerver_userstatus",
    "zerver_usertopic",
    "zerver_muteduser",
}

# This set contains those database tables that we expect to not be
# included in the export.  This tool does validation to ensure that
# every table in the database is either exported or listed here, to
# ensure we never accidentally fail to export a table.
NON_EXPORTED_TABLES = {
    # These invitation/confirmation flow tables don't make sense to
    # export, since invitations links will be broken by the server URL
    # change anyway:
    "zerver_emailchangestatus",
    "zerver_multiuseinvite",
    "zerver_multiuseinvite_streams",
    "zerver_multiuseinvite_groups",
    "zerver_preregistrationrealm",
    "zerver_preregistrationuser",
    "zerver_preregistrationuser_streams",
    "zerver_preregistrationuser_groups",
    "zerver_realmreactivationstatus",
    "zerver_realmcreationstatus",
    # Missed message addresses are low value to export since
    # missed-message email addresses include the server's hostname and
    # expire after a few days.
    "zerver_missedmessageemailaddress",
    # Scheduled message notification email data is for internal use by the server.
    "zerver_scheduledmessagenotificationemail",
    # When switching servers, clients will need to re-log in and
    # reregister for push notifications anyway.
    "zerver_pushdevice",
    "zerver_pushdevicetoken",
    # We don't use these generated Django tables
    "zerver_userprofile_groups",
    "zerver_userprofile_user_permissions",
    # These is used for scheduling future activity; it could make
    # sense to export, but is relatively low value.
    "zerver_scheduledemail",
    "zerver_scheduledemail_users",
    # These tables are related to a user's 2FA authentication
    # configuration, which will need to be set up again on the new
    # server.
    "two_factor_phonedevice",
    "otp_static_staticdevice",
    "otp_static_statictoken",
    "otp_totp_totpdevice",
    # These archive tables should not be exported (they are to support
    # restoring content accidentally deleted due to software bugs in
    # the retention policy feature)
    "zerver_archivedmessage",
    "zerver_archivedusermessage",
    "zerver_archivedattachment",
    "zerver_archivedattachment_messages",
    "zerver_archivedreaction",
    "zerver_archivedsubmessage",
    "zerver_archivetransaction",
    # Social auth tables are not needed post-export, since we don't
    # use any of this state outside of a direct authentication flow.
    "social_auth_association",
    "social_auth_code",
    "social_auth_nonce",
    "social_auth_partial",
    "social_auth_usersocialauth",
    # We will likely never want to migrate this table, since it's a
    # total of all the realmcount values on the server.  Might need to
    # recompute it after a fillstate import.
    "analytics_installationcount",
    # Fillstate will require some cleverness to do the right partial export.
    "analytics_fillstate",
    # These are for unfinished features; we'll want to add them to the
    # export before they reach full production status.
    "zerver_defaultstreamgroup",
    "zerver_defaultstreamgroup_streams",
    # Drafts don't need to be exported as they are supposed to be more ephemeral.
    "zerver_draft",
    # The importer cannot trust ImageAttachment objects anyway and needs to check
    # and process images for thumbnailing on its own.
    "zerver_imageattachment",
    # ChannelEmailAddress entries are low value to export since
    # channel email addresses include the server's hostname.
    "zerver_channelemailaddress",
    # For any tables listed below here, it's a bug that they are not present in the export.
}

IMPLICIT_TABLES = {
    # ManyToMany relationships are exported implicitly when importing
    # the parent table.
    "zerver_attachment_messages",
    "zerver_attachment_scheduled_messages",
}

ATTACHMENT_TABLES = {
    "zerver_attachment",
}

MESSAGE_TABLES = {
    # message tables get special treatment, because they're by far our
    # largest tables and need to be paginated.
    "zerver_message",
    "zerver_usermessage",
    # zerver_reaction and zerver_submessage are effectively metadata
    # attached to messages that are used to display the message.
    "zerver_reaction",
    "zerver_submessage",
    # zerver_client is also written after we know what clients got
    # used
    "zerver_client",
}

# These get their own file as analytics data can be quite large and
# would otherwise make realm.json unpleasant to manually inspect
ANALYTICS_TABLES = {
    "analytics_realmcount",
    "analytics_streamcount",
    "analytics_usercount",
}

# This data structure lists all the Django DateTimeField fields in the
# data model.  These are converted to floats during the export process
# via floatify_datetime_fields, and back during the import process.
#
# TODO: This data structure could likely eventually be replaced by
# inspecting the corresponding Django models
DATE_FIELDS: dict[TableName, list[Field]] = {
    "analytics_installationcount": ["end_time"],
    "analytics_realmcount": ["end_time"],
    "analytics_streamcount": ["end_time"],
    "analytics_usercount": ["end_time"],
    "zerver_attachment": ["create_time"],
    "zerver_channelfolder": ["date_created"],
    "zerver_externalauthid": ["date_created"],
    "zerver_message": ["last_edit_time", "date_sent"],
    "zerver_muteduser": ["date_muted"],
    "zerver_realmauditlog": ["event_time"],
    "zerver_realm": [
        "date_created",
        "demo_organization_scheduled_deletion_date",
        "push_notifications_enabled_end_timestamp",
        "scheduled_deletion_date",
    ],
    "zerver_realmexport": [
        "date_requested",
        "date_started",
        "date_succeeded",
        "date_failed",
        "date_deleted",
    ],
    "zerver_savedsnippet": ["date_created"],
    "zerver_scheduledmessage": ["scheduled_timestamp", "request_timestamp"],
    "zerver_stream": ["date_created"],
    "zerver_namedusergroup": ["date_created"],
    "zerver_useractivityinterval": ["start", "end"],
    "zerver_useractivity": ["last_visit"],
    "zerver_onboardingstep": ["timestamp"],
    "zerver_userpresence": ["last_active_time", "last_connected_time"],
    "zerver_userprofile": ["date_joined", "last_login", "last_reminder"],
    "zerver_userprofile_mirrordummy": ["date_joined", "last_login", "last_reminder"],
    "zerver_userstatus": ["timestamp"],
    "zerver_usertopic": ["last_updated"],
}


def sanity_check_output(data: TableData) -> None:
    # First, we verify that the export tool has a declared
    # configuration for every table declared in the `models` modules.
    target_models = [
        *apps.get_app_config("analytics").get_models(include_auto_created=True),
        *apps.get_app_config("django_otp").get_models(include_auto_created=True),
        *apps.get_app_config("otp_static").get_models(include_auto_created=True),
        *apps.get_app_config("otp_totp").get_models(include_auto_created=True),
        *apps.get_app_config("phonenumber").get_models(include_auto_created=True),
        *apps.get_app_config("social_django").get_models(include_auto_created=True),
        *apps.get_app_config("two_factor").get_models(include_auto_created=True),
        *apps.get_app_config("zerver").get_models(include_auto_created=True),
    ]
    all_tables_db = {model._meta.db_table for model in target_models}

    # These assertion statements will fire when we add a new database
    # table that is not included in Zulip's data exports.  Generally,
    # you can add your new table to `ALL_ZULIP_TABLES` and
    # `NON_EXPORTED_TABLES` during early work on a new feature so that
    # CI passes.
    #
    # We'll want to make sure we handle it for exports before
    # releasing the new feature, but doing so correctly requires some
    # expertise on this export system.
    error_message = f"""
    It appears you've added a new database table, but haven't yet
    registered it in ALL_ZULIP_TABLES and the related declarations
    in {__file__} for what to include in data exports.
    """

    assert all_tables_db == ALL_ZULIP_TABLES, error_message
    assert NON_EXPORTED_TABLES.issubset(ALL_ZULIP_TABLES), error_message
    assert IMPLICIT_TABLES.issubset(ALL_ZULIP_TABLES), error_message
    assert ATTACHMENT_TABLES.issubset(ALL_ZULIP_TABLES), error_message
    assert ANALYTICS_TABLES.issubset(ALL_ZULIP_TABLES), error_message

    tables = set(ALL_ZULIP_TABLES)
    tables -= NON_EXPORTED_TABLES
    tables -= IMPLICIT_TABLES
    tables -= MESSAGE_TABLES
    tables -= ATTACHMENT_TABLES
    tables -= ANALYTICS_TABLES

    for table in tables:
        if table not in data:
            logging.warning("??? NO DATA EXPORTED FOR TABLE %s!!!", table)


def orjson_stream(
    it: Iterable[Any],
    options: int = orjson.OPT_INDENT_2,
    indent: bytes = b"",
    chunk_size: int = ORJSON_ITERABLE_BATCH_SIZE,
) -> Iterator[bytes]:
    first_chunk = True
    for batch in batched(it, chunk_size):
        if not first_chunk:
            yield b",\n" + indent
        chunk = orjson.dumps(batch, option=options)
        if not first_chunk:
            assert chunk.startswith(b"[\n")
            chunk = chunk[2:]
        assert chunk.endswith(b"\n]")
        chunk = chunk[:-2]
        if indent != b"":
            chunk = chunk.replace(b"\n", b"\n" + indent)
        yield chunk
        first_chunk = False
    if first_chunk:
        yield b"[]"
    else:
        yield b"\n" + indent + b"]"


def orjson_serialize_iterable(
    obj: Any, options: int = orjson.OPT_INDENT_2, indent: bytes = b""
) -> orjson.Fragment:
    if not isinstance(obj, Iterator):
        raise TypeError
    serialized = bytearray()
    for byte_section in orjson_stream(obj, options, indent):
        serialized.extend(byte_section)
    return orjson.Fragment(bytes(serialized))


def write_data_to_file(output_file: Path, data: Any) -> None:
    """
    IMPORTANT: You generally don't want to call this directly.

    Instead use one of the higher level helpers:

        write_table_data
        write_records_json_file

    The one place we call this directly is for message partials.
    """
    with open(output_file, "wb") as f:
        # Because we don't pass a default handler, OPT_PASSTHROUGH_DATETIME
        # actually causes orjson to raise a TypeError on datetime objects. This
        # is what we want, because it helps us check that we correctly
        # post-processed them to serialize to UNIX timestamps rather than ISO
        # 8601 strings for historical reasons.
        options = orjson.OPT_INDENT_2 | orjson.OPT_PASSTHROUGH_DATETIME
        f.write(
            orjson.dumps(
                data,
                option=options,
                default=lambda d: orjson_serialize_iterable(d, options, indent=b"  "),
            )
        )
    logging.info("Finished writing %s", output_file)


def write_table_data(output_file: str, data: dict[str, Any]) -> None:
    # We sort by ids mostly so that humans can quickly do diffs
    # on two export jobs to see what changed (either due to new
    # data arriving or new code being deployed).
    for value in data.values():
        if isinstance(value, list):
            value.sort(key=lambda row: row["id"])

    assert output_file.endswith(".json")

    write_data_to_file(output_file, data)


def write_records_json_file(output_dir: str, records: Iterable[dict[str, Any]]) -> None:
    # We want a somewhat deterministic sorting order here. All of our
    # versions of records.json include a "path" field in each element,
    # even though there's some variation among avatars/emoji/realm_icons/uploads
    # in other fields that get written.
    #
    # The sorting order of paths isn't entirely sensical to humans,
    # because they include ids and even some random numbers,
    # but if you export the same realm twice, you should get identical results.
    #
    # If this is an iterator, it is promised to be already sorted.
    # For lists, we sort by that here.
    if isinstance(records, list):
        records.sort(key=lambda record: record["path"])

    output_file = os.path.join(output_dir, "records.json")
    with open(output_file, "wb") as f:
        # For legacy reasons we allow datetime objects here, unlike
        # write_data_to_file.
        f.writelines(orjson_stream(records))
    logging.info("Finished writing %s", output_file)


def make_raw(query: Iterable[Any], exclude: list[Field] | None = None) -> Iterator[Record]:
    """
    Takes a Django query and returns a JSONable list
    of dictionaries corresponding to the database rows.
    """
    for instance in query:
        data = model_to_dict(instance, exclude=exclude)
        """
        In Django 1.11.5, model_to_dict evaluates the QuerySet of
        many-to-many field to give us a list of instances. We require
        a list of primary keys, so we get the primary keys from the
        instances below.
        """
        for field in instance._meta.many_to_many:
            if exclude is not None and field.name in exclude:
                continue
            value = data[field.name]
            data[field.name] = [row.id for row in value]

        yield data


def floatify_datetime_fields(item: Record, table: TableName) -> Record:
    updates = {}
    for field in DATE_FIELDS[table]:
        dt = item[field]
        if dt is None:
            continue
        assert isinstance(dt, datetime)
        assert not timezone_is_naive(dt)
        updates[field] = dt.timestamp()
    return {**item, **updates}


class Config:
    """A Config object configures a single table for exporting (and, maybe
    some day importing as well.  This configuration defines what
    process needs to be followed to correctly extract the set of
    objects to export.

    You should never mutate Config objects as part of the export;
    instead use the data to determine how you populate other
    data structures.

    There are parent/children relationships between Config objects.
    The parent should be instantiated first.  The child will
    append itself to the parent's list of children.

    """

    def __init__(
        self,
        table: str | None = None,
        model: Any | None = None,
        normal_parent: Optional["Config"] = None,
        virtual_parent: Optional["Config"] = None,
        filter_args: FilterArgs | None = None,
        custom_fetch: CustomFetch | None = None,
        custom_tables: list[TableName] | None = None,
        custom_return_ids: CustomReturnIds | None = None,
        custom_process_results: CustomProcessResults | None = None,
        concat_and_destroy: list[TableName] | None = None,
        id_source: IdSource | None = None,
        source_filter: SourceFilter | None = None,
        include_rows: Field | None = None,
        is_seeded: bool = False,
        exclude: list[Field] | None = None,
        limit_to_consenting_users: bool | None = None,
        collect_client_ids: bool = False,
        use_iterator: bool = True,
    ) -> None:
        assert table or custom_tables
        self.table = table
        self.model = model
        self.normal_parent = normal_parent
        self.virtual_parent = virtual_parent
        self.filter_args = filter_args
        self.include_rows = include_rows
        self.is_seeded = is_seeded
        self.exclude = exclude
        self.custom_fetch = custom_fetch
        self.custom_tables = custom_tables
        self.custom_return_ids = custom_return_ids
        self.custom_process_results = custom_process_results
        self.concat_and_destroy = concat_and_destroy
        self.id_source = id_source
        self.source_filter = source_filter
        self.limit_to_consenting_users = limit_to_consenting_users
        self.collect_client_ids = collect_client_ids
        self.children: list[Config] = []

        if self.include_rows:
            assert self.include_rows.endswith("_id__in")

        if self.custom_fetch:
            # enforce a naming convention
            assert self.custom_fetch.__name__.startswith("custom_fetch_")
            if self.normal_parent is not None:
                raise AssertionError(
                    """
                    If you have a custom fetcher, then specify
                    your parent as a virtual_parent.
                    """
                )

            if self.collect_client_ids:
                raise AssertionError(
                    """
                    If you're using custom_fetch with collect_client_ids, you need to
                    extend the related logic to handle how to collect Client ids with your
                    customer fetcher.
                    """
                )

        if normal_parent is not None:
            self.parent: Config | None = normal_parent
        else:
            self.parent = None

        if virtual_parent is not None and normal_parent is not None:
            raise AssertionError(
                """
                If you specify a normal_parent, please
                do not create a virtual_parent.
                """
            )

        if normal_parent is not None:
            normal_parent.children.append(self)
        elif virtual_parent is not None:
            virtual_parent.children.append(self)
        elif is_seeded is None:
            raise AssertionError(
                """
                You must specify a parent if you are
                not using is_seeded.
                """
            )

        if (
            (parent := normal_parent or virtual_parent) is not None
            and parent.table == "zerver_userprofile"
            and limit_to_consenting_users is None
        ):
            raise AssertionError(
                """
                Config having UserProfile as a parent must pass limit_to_consenting_users
                explicitly.
                """
            )

        if self.id_source is not None:
            if self.virtual_parent is None:
                raise AssertionError(
                    """
                    You must specify a virtual_parent if you are
                    using id_source."""
                )
            if self.id_source[0] != self.virtual_parent.table:
                raise AssertionError(
                    f"""
                    Configuration error.  To populate {self.table}, you
                    want data from {self.id_source[0]}, but that differs from
                    the table name of your virtual parent ({self.virtual_parent.table}),
                    which suggests you many not have set up
                    the ordering correctly.  You may simply
                    need to assign a virtual_parent, or there
                    may be deeper issues going on."""
                )

        if self.limit_to_consenting_users:
            # Combining these makes no sense. limit_to_consenting_users is used to restrict queries
            # for Configs which use include_rows="user_profile_id__in" to only pass user ids of users
            # who have consented to private data export.
            # If a Config defines its own custom_fetch, then it is fully responsible for doing its own
            # queries - so it doesn't integrate with limit_to_consenting_users.
            assert not self.custom_fetch

            assert include_rows in ["user_profile_id__in", "user_id__in", "bot_profile_id__in"]
            assert normal_parent is not None and normal_parent.table == "zerver_userprofile"

        if self.collect_client_ids or self.is_seeded:
            self.use_iterator = False
        else:
            self.use_iterator = use_iterator

    def return_ids(self, response: TableData) -> set[int]:
        if self.custom_return_ids is not None:
            return self.custom_return_ids(response)
        else:
            assert self.table is not None
            assert not self.use_iterator, self.table
            return {row["id"] for row in response[self.table]}


def export_from_config(
    response: TableData,
    config: Config,
    seed_object: Any | None = None,
    context: Context | None = None,
) -> None:
    table = config.table
    parent = config.parent
    model = config.model

    if context is None:
        context = {}

    if config.custom_tables:
        exported_tables = config.custom_tables
    else:
        assert table is not None, """
            You must specify config.custom_tables if you
            are not specifying config.table"""
        exported_tables = [table]

    for t in exported_tables:
        logging.info("Exporting via export_from_config:  %s", t)

    rows: Iterable[Any] | None = None
    query: QuerySet[Any] | None = None
    if config.is_seeded:
        rows = [seed_object]

    elif config.custom_fetch:
        config.custom_fetch(
            response,
            context,
        )
        if config.custom_tables:
            for t in config.custom_tables:
                if t not in response:
                    raise AssertionError(f"Custom fetch failed to populate {t}")

    elif config.concat_and_destroy:
        # When we concat_and_destroy, we are working with
        # temporary "tables" that are lists of records that
        # should already be ready to export.
        assert table is not None
        # We pop them off of the response and store them in a local
        # which the iterable closes over
        tables = {t: response.pop(t) for t in config.concat_and_destroy}
        response[table] = chain.from_iterable(tables[t] for t in config.concat_and_destroy)
        for t in config.concat_and_destroy:
            logging.info("Deleted temporary %s", t)

    elif config.normal_parent:
        # In this mode, our current model is figuratively Article,
        # and normal_parent is figuratively Blog, and
        # now we just need to get all the articles
        # contained by the blogs.
        model = config.model
        assert parent is not None
        assert parent.table is not None
        assert config.include_rows is not None
        parent_ids = parent.return_ids(response)
        filter_params: dict[str, object] = {config.include_rows: parent_ids}

        if config.filter_args is not None:
            filter_params.update(config.filter_args)
        if config.limit_to_consenting_users:
            if "realm" in context:
                realm = context["realm"]
                export_type = context["export_type"]
                assert isinstance(realm, Realm)
                if export_type == RealmExport.EXPORT_PUBLIC:
                    # In a public export, no private data is exported, so
                    # no users are considered consenting.
                    consenting_user_ids: set[int] | None = set()
                elif export_type == RealmExport.EXPORT_FULL_WITH_CONSENT:
                    consenting_user_ids = context["exportable_user_ids"]
                else:
                    assert export_type == RealmExport.EXPORT_FULL_WITHOUT_CONSENT
                    # In a full export without consent, the concept is meaningless,
                    # so set this to None. All private data will be exported without consulting
                    # consenting_user_ids so we set this to None so that any code in this flow
                    # which (incorrectly) tries to access them fails explicitly.
                    consenting_user_ids = None
            else:
                # Single user export. This should not be really relevant, because
                # limit_to_consenting_users is unlikely to be used in Configs in that codepath,
                # as we should be exporting only a single user's data anyway; but it's still
                # useful to have this case written correctly for robustness.
                assert "user" in context
                assert isinstance(context["user"], UserProfile)
                export_type = None
                consenting_user_ids = {context["user"].id}

            user_profile_id_in_key = config.include_rows

            # Sanity check.
            assert user_profile_id_in_key in [
                "user_profile_id__in",
                "user_id__in",
                "bot_profile_id__in",
            ]

            user_profile_id_in = filter_params[user_profile_id_in_key]
            assert isinstance(user_profile_id_in, set)

            if export_type != RealmExport.EXPORT_FULL_WITHOUT_CONSENT:
                assert consenting_user_ids is not None
                filter_params[user_profile_id_in_key] = consenting_user_ids.intersection(
                    user_profile_id_in
                )

        assert model is not None
        try:
            query = model.objects.filter(**filter_params).order_by("id")
        except Exception:
            print(
                f"""
                Something about your Config seems to make it difficult
                to construct a query.

                table: {table}
                parent: {parent.table}

                filter_params: {filter_params}
                """
            )
            raise

    elif config.id_source:
        # In this mode, we are the figurative Blog, and we now
        # need to look at the current response to get all the
        # blog ids from the Article rows we fetched previously.
        model = config.model
        assert model is not None
        # This will be a tuple of the form ('zerver_article', 'blog').
        (child_table, field) = config.id_source
        assert config.virtual_parent is not None
        assert not config.virtual_parent.use_iterator
        child_rows = response[child_table]
        if config.source_filter:
            child_rows = [r for r in child_rows if config.source_filter(r)]
        lookup_ids = [r[field] for r in child_rows]
        filter_params = dict(id__in=lookup_ids)
        if config.filter_args:
            filter_params.update(config.filter_args)
        query = model.objects.filter(**filter_params)

    if query is not None:
        rows = query.iterator()

