def two_antidepressant_episodes(
        patient_drug_date_df: DataFrame,
        patient_colname: str = DEFAULT_SOURCE_PATIENT_COLNAME,
        drug_colname: str = DEFAULT_SOURCE_DRUG_COLNAME,
        date_colname: str = DEFAULT_SOURCE_DATE_COLNAME,
        course_length_days: int = DEFAULT_ANTIDEPRESSANT_COURSE_LENGTH_DAYS,
        expect_response_by_days: int = DEFAULT_EXPECT_RESPONSE_BY_DAYS,
        symptom_assessment_time_days: int =
        DEFAULT_SYMPTOM_ASSESSMENT_TIME_DAYS,
        n_threads: int = DEFAULT_N_THREADS,
        first_episode_only: bool = True) -> DataFrame:
    """
    Takes a *pandas* ``DataFrame``, ``patient_drug_date_df`` (or, via
    ``reticulate``, an R ``data.frame`` or ``data.table``). This should contain
    dated present-tense references to antidepressant drugs (only).

    Returns a set of result rows as a ``DataFrame``.
    """
    # Say hello
    log.info("Running two_antidepressant_episodes...")
    start = Pendulum.now()

    # Work through each patient
    patient_ids = sorted(list(set(patient_drug_date_df[patient_colname])))
    n_patients = len(patient_ids)
    log.info("Found {} patients", n_patients)
    flush_stdout_stderr()

    def _get_patient_result(_patient_id: str) -> Optional[DataFrame]:
        return two_antidepressant_episodes_single_patient(
            patient_id=_patient_id,
            patient_drug_date_df=patient_drug_date_df,
            patient_colname=patient_colname,
            drug_colname=drug_colname,
            date_colname=date_colname,
            course_length_days=course_length_days,
            expect_response_by_days=expect_response_by_days,
            symptom_assessment_time_days=symptom_assessment_time_days,
            first_episode_only=first_episode_only,
        )

    combined_result = _get_blank_two_antidep_episodes_result()
    if n_threads > 1:
        # Farm off the work to lots of threads:
        log.info("Parallel processing method; {} threads", n_threads)
        with ThreadPoolExecutor(max_workers=n_threads) as executor:
            list_of_results_frames = executor.map(_get_patient_result,
                                                  patient_ids)

        log.debug("Recombining results from parallel processing...")
        for patient_result in list_of_results_frames:
            if patient_result is not None:
                combined_result = combined_result.append(patient_result)
        log.debug("... recombined")

    else:
        log.info("Single-thread method")
        for ptnum, patient_id in enumerate(patient_ids, start=1):
            log.debug("Processing patient {} out of {}", ptnum, n_patients)
            patient_result = _get_patient_result(patient_id)
            if patient_result is not None:
                combined_result = combined_result.append(patient_result)

    # For consistent results order, even with parallel processing:
    combined_result = combined_result.sort_values(
        by=[RCN_PATIENT_ID], ascending=True)
    # So that the DataFrame indices aren't all zero (largely cosmetic):
    combined_result.reset_index(inplace=True, drop=True)

    end = Pendulum.now()
    duration = end - start
    log.info("Took {} seconds for {} patients",
             duration.total_seconds(), n_patients)
    flush_stdout_stderr()
    return combined_result