# Path: pemi/data_subject.py
# class PdDataSubject(DataSubject):
#     def __init__(self, df=None, strict_match_schema=False, **kwargs):
#         super().__init__(**kwargs)
# 
#         if df is None or df.shape == (0, 0):
#             df = self._empty_df()
#         self.strict_match_schema = strict_match_schema
#         self.df = df
# 
#     def to_pd(self):
#         return self.df
# 
#     def from_pd(self, df, **kwargs):
#         self.df = df
# 
#     def connect_from(self, other):
#         if other.df is None or other.df.shape == (0, 0):
#             self.df = self._empty_df()
#         else:
#             self.df = other.df
#         self.validate_schema()
# 
#     def validate_schema(self):
#         'Verify that the dataframe contains all of the columns specified in the schema'
#         if self.strict_match_schema:
#             return self.validate_data_frame_columns()
#         missing = set(self.schema.keys()) - set(self.df.columns)
#         if len(missing) == 0:
#             return True
#         raise MissingFieldsError('DataFrame missing expected fields: {}'.format(missing))
# 
#     def validate_data_frame_columns(self):
#         'Verify that the schema contains all the columns specefied in the dataframe'
#         missing = set(self.df.columns) - set(self.schema.keys())
#         if len(missing) > 0:
#             raise MissingFieldsError("Schema is missing current columns: {}".format(missing))
#         return True
# 
#     def _empty_df(self):
#         return pd.DataFrame(columns=self.schema.keys())


import re
import sys
import pandas as pd
import factory
import pemi
import pemi.testing as pt
from collections import OrderedDict
from pemi.data_subject import PdDataSubject
from pemi.fields import *
    ),
    'beers_w_style': pemi.Schema(
        id=IntegerField(),
        name=StringField(),
        abv=DecimalField(precision=3, scale=1),
        style=StringField()
    )
}

class RemoteSourcePipe(pemi.Pipe):
    def __init__(self, **params):
        super().__init__(**params)

        self.target(
            PdDataSubject,
            name='main',
            schema=this.schemas['beers']
        )

    def flow(self):
        pemi.log().debug('FLOWING {}'.format(self))
        self.targets['main'].df = pd.DataFrame({'id': [1, 2, 3], 'name': ['one', 'two', 'three']})
        raise NotImplementedError('The test needs to mock out external calls')


class RemoteTargetPipe(pemi.Pipe):
    def __init__(self, **params):
        super().__init__(**params)

        self.source(

# Path: pemi/data_subject.py
# class PdDataSubject(DataSubject):
#     def __init__(self, df=None, strict_match_schema=False, **kwargs):
#         super().__init__(**kwargs)
# 
#         if df is None or df.shape == (0, 0):
#             df = self._empty_df()
#         self.strict_match_schema = strict_match_schema
#         self.df = df
# 
#     def to_pd(self):
#         return self.df
# 
#     def from_pd(self, df, **kwargs):
#         self.df = df
# 
#     def connect_from(self, other):
#         if other.df is None or other.df.shape == (0, 0):
#             self.df = self._empty_df()
#         else:
#             self.df = other.df
#         self.validate_schema()
# 
#     def validate_schema(self):
#         'Verify that the dataframe contains all of the columns specified in the schema'
#         if self.strict_match_schema:
#             return self.validate_data_frame_columns()
#         missing = set(self.schema.keys()) - set(self.df.columns)
#         if len(missing) == 0:
#             return True
#         raise MissingFieldsError('DataFrame missing expected fields: {}'.format(missing))
# 
#     def validate_data_frame_columns(self):
#         'Verify that the schema contains all the columns specefied in the dataframe'
#         missing = set(self.df.columns) - set(self.schema.keys())
#         if len(missing) > 0:
#             raise MissingFieldsError("Schema is missing current columns: {}".format(missing))
#         return True
# 
#     def _empty_df(self):
#         return pd.DataFrame(columns=self.schema.keys())


# Path: tests/test_job.py
import re
import sys
import pandas as pd
import factory
import pemi
import pemi.testing as pt
from collections import OrderedDict
from pemi.data_subject import PdDataSubject
from pemi.fields import *

    ),
    'beers_w_style': pemi.Schema(
        id=IntegerField(),
        name=StringField(),
        abv=DecimalField(precision=3, scale=1),
        style=StringField()
    )
}

class RemoteSourcePipe(pemi.Pipe):
    def __init__(self, **params):
        super().__init__(**params)

        self.target(
            PdDataSubject,
            name='main',
            schema=this.schemas['beers']
        )

    def flow(self):
        pemi.log().debug('FLOWING {}'.format(self))
        self.targets['main'].df = pd.DataFrame({'id': [1, 2, 3], 'name': ['one', 'two', 'three']})
        raise NotImplementedError('The test needs to mock out external calls')


class RemoteTargetPipe(pemi.Pipe):
    def __init__(self, **params):
        super().__init__(**params)

        self.source(

            PdDataSubject,