# Path: pemi/data_subject.py
# class SaDataSubject(DataSubject):
#     def __init__(self, engine, table, sql_schema=None, **kwargs):
#         super().__init__(**kwargs)
#         self.engine = engine
#         self.table = table
#         self.sql_schema = sql_schema
# 
#         self.cached_test_df = None
# 
#     def to_pd(self):
#         if self.cached_test_df is not None:
#             return self.cached_test_df
# 
#         with self.engine.connect() as conn:
#             df = pd.read_sql_table(
#                 self.table,
#                 conn,
#                 schema=self.sql_schema,
#             )
# 
#             for column in set(df.columns) & set(self.schema.keys()):
#                 df[column] = df[column].apply(self.schema[column].coerce)
# 
#             self.cached_test_df = df
#         return df
# 
#     def from_pd(self, df, **to_sql_opts):
#         self.cached_test_df = df
#         pemi.log.debug('loading SaDataSubject with:\n%s', self.cached_test_df)
# 
#         to_sql_opts['if_exists'] = to_sql_opts.get('if_exists', 'append')
#         to_sql_opts['index'] = to_sql_opts.get('index', False)
#         if self.sql_schema:
#             to_sql_opts['schema'] = self.sql_schema
# 
#         df_to_sql = df.copy()
#         for field in self.schema.values():
#             if isinstance(field, JsonField):
#                 df_to_sql[field.name] = df_to_sql[field.name].apply(json.dumps)
# 
#         with self.engine.connect() as conn:
#             df_to_sql.to_sql(self.table, conn, **to_sql_opts)
# 
#     def connect_from(self, _other):
#         self.engine.dispose()
#         self.validate_schema()
# 
#     def __getstate__(self):
#         return (
#             [],
#             {
#                 'url': self.engine.url,
#                 'table': self.table,
#                 'sql_schema': self.sql_schema
#             }
#         )
# 
#     def __setstate__(self, state):
#         _args, kwargs = state
#         self.engine = sa.create_engine(kwargs['url'])
#         self.table = kwargs['table']
#         self.sql_schema = kwargs['sql_schema']


import os
import logging
import factory
import sqlalchemy as sa
import dask.threaded
import pemi
import pemi.testing as pt
from pemi.data_subject import SaDataSubject
from pemi.fields import *
            name='renamed_beers',
            pipe=RenameSaPipe(
                schema=SCHEMAS['beers'],
                table='beers'
            )
        )
        self.connect('renamed_beers', 'main').to('denormalizer', 'beers')

        self.pipe(
            name='denormalizer',
            pipe=DenormalizeBeersPipe()
        )

        self.target(
            SaDataSubject,
            name='beer_sales',
            schema=SCHEMAS['beer_sales'],
            engine=SA_ENGINE,
            table='beer_sales'
        )

    def flow(self):
        self.connections.flow(dask_get=dask.threaded.get)


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

        self.source(

# Path: pemi/data_subject.py
# class SaDataSubject(DataSubject):
#     def __init__(self, engine, table, sql_schema=None, **kwargs):
#         super().__init__(**kwargs)
#         self.engine = engine
#         self.table = table
#         self.sql_schema = sql_schema
# 
#         self.cached_test_df = None
# 
#     def to_pd(self):
#         if self.cached_test_df is not None:
#             return self.cached_test_df
# 
#         with self.engine.connect() as conn:
#             df = pd.read_sql_table(
#                 self.table,
#                 conn,
#                 schema=self.sql_schema,
#             )
# 
#             for column in set(df.columns) & set(self.schema.keys()):
#                 df[column] = df[column].apply(self.schema[column].coerce)
# 
#             self.cached_test_df = df
#         return df
# 
#     def from_pd(self, df, **to_sql_opts):
#         self.cached_test_df = df
#         pemi.log.debug('loading SaDataSubject with:\n%s', self.cached_test_df)
# 
#         to_sql_opts['if_exists'] = to_sql_opts.get('if_exists', 'append')
#         to_sql_opts['index'] = to_sql_opts.get('index', False)
#         if self.sql_schema:
#             to_sql_opts['schema'] = self.sql_schema
# 
#         df_to_sql = df.copy()
#         for field in self.schema.values():
#             if isinstance(field, JsonField):
#                 df_to_sql[field.name] = df_to_sql[field.name].apply(json.dumps)
# 
#         with self.engine.connect() as conn:
#             df_to_sql.to_sql(self.table, conn, **to_sql_opts)
# 
#     def connect_from(self, _other):
#         self.engine.dispose()
#         self.validate_schema()
# 
#     def __getstate__(self):
#         return (
#             [],
#             {
#                 'url': self.engine.url,
#                 'table': self.table,
#                 'sql_schema': self.sql_schema
#             }
#         )
# 
#     def __setstate__(self, state):
#         _args, kwargs = state
#         self.engine = sa.create_engine(kwargs['url'])
#         self.table = kwargs['table']
#         self.sql_schema = kwargs['sql_schema']


# Path: tests/test_db_dask_job.py
import os
import logging
import factory
import sqlalchemy as sa
import dask.threaded
import pemi
import pemi.testing as pt
from pemi.data_subject import SaDataSubject
from pemi.fields import *

            name='renamed_beers',
            pipe=RenameSaPipe(
                schema=SCHEMAS['beers'],
                table='beers'
            )
        )
        self.connect('renamed_beers', 'main').to('denormalizer', 'beers')

        self.pipe(
            name='denormalizer',
            pipe=DenormalizeBeersPipe()
        )

        self.target(
            SaDataSubject,
            name='beer_sales',
            schema=SCHEMAS['beer_sales'],
            engine=SA_ENGINE,
            table='beer_sales'
        )

    def flow(self):
        self.connections.flow(dask_get=dask.threaded.get)


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

        self.source(

            SaDataSubject,