# Path: pemi/pipes/patterns.py
# class TargetPipe(pemi.Pipe):
#     '''
#     A target pipe takes data provided to it, encodes it into a structure that can be
#     understand by some external target, and then loads the data into that external target.
#     '''
# 
#     def __init__(self, *, schema, **params):
#         super().__init__(**params)
# 
#         self.schema = schema
# 
#         self.source(
#             pemi.PdDataSubject,
#             name='main',
#             schema=self.schema
#         )
# 
#         self.target(
#             pemi.PdDataSubject,
#             name='errors'
#         )
# 
#         self.target(
#             pemi.PdDataSubject,
#             name='response'
#         )
# 
#     def encode(self):
#         # e.g., CsvTargetEncoder.encode()
#         # return source_main_data_encoded_for_loader
#         raise NotImplementedError
# 
#     def load(self, encoded_data):
#         # e.g., S3Loader.load()
#         # return results_from_load_operation
#         raise NotImplementedError
# 
#     def flow(self):
#         self.load(self.encode())


import os
import re
import pandas as pd
import pemi
from pemi.pipes.patterns import TargetPipe
            'error_bad_lines': True
        }

        return {**default_opts, **user_csv_opts, **mandatory_opts}

    def _parse_one(self, filepath):
        pemi.log.debug('Parsing file at %s', filepath)

        raw_df = pd.read_csv(filepath, **self.csv_opts)
        pemi.log.debug('Found %i raw records', len(raw_df))

        raw_df.columns = [self.column_normalizer(col) for col in raw_df.columns]

        if self.filename_field:
            if self.filename_full_path:
                raw_df[self.filename_field] = filepath
            else:
                raw_df[self.filename_field] = os.path.basename(filepath)

        if self.schema:
            return raw_df.mapping(
                [(name, name, field.coerce) for name, field in self.schema.items()],
                on_error='redirect'
            )
        return raw_df.mapping([], inplace=True)

    def flow(self):
        self.parse(self.extract())



# Path: pemi/pipes/patterns.py
# class TargetPipe(pemi.Pipe):
#     '''
#     A target pipe takes data provided to it, encodes it into a structure that can be
#     understand by some external target, and then loads the data into that external target.
#     '''
# 
#     def __init__(self, *, schema, **params):
#         super().__init__(**params)
# 
#         self.schema = schema
# 
#         self.source(
#             pemi.PdDataSubject,
#             name='main',
#             schema=self.schema
#         )
# 
#         self.target(
#             pemi.PdDataSubject,
#             name='errors'
#         )
# 
#         self.target(
#             pemi.PdDataSubject,
#             name='response'
#         )
# 
#     def encode(self):
#         # e.g., CsvTargetEncoder.encode()
#         # return source_main_data_encoded_for_loader
#         raise NotImplementedError
# 
#     def load(self, encoded_data):
#         # e.g., S3Loader.load()
#         # return results_from_load_operation
#         raise NotImplementedError
# 
#     def flow(self):
#         self.load(self.encode())


# Path: pemi/pipes/csv.py
import os
import re
import pandas as pd
import pemi
from pemi.pipes.patterns import TargetPipe

            'error_bad_lines': True
        }

        return {**default_opts, **user_csv_opts, **mandatory_opts}

    def _parse_one(self, filepath):
        pemi.log.debug('Parsing file at %s', filepath)

        raw_df = pd.read_csv(filepath, **self.csv_opts)
        pemi.log.debug('Found %i raw records', len(raw_df))

        raw_df.columns = [self.column_normalizer(col) for col in raw_df.columns]

        if self.filename_field:
            if self.filename_full_path:
                raw_df[self.filename_field] = filepath
            else:
                raw_df[self.filename_field] = os.path.basename(filepath)

        if self.schema:
            return raw_df.mapping(
                [(name, name, field.coerce) for name, field in self.schema.items()],
                on_error='redirect'
            )
        return raw_df.mapping([], inplace=True)

    def flow(self):
        self.parse(self.extract())



class LocalCsvFileTargetPipe(TargetPipe):