https://dagster.io/ logo
#ask-community
Title
# ask-community
b

Brendan Jackson

06/23/2023, 1:39 PM
Hi, I'm using IOManagers that depend on other resources. I have one
DateIntervalSqlIOManager
that depends on
sql_resource
and implements
setup_for_execution
, and works fine. However, the moment I implement
teardown_after_execution
, I get the following error:
Copy code
dagster._core.errors.DagsterSubprocessError: During multiprocess execution errors occurred in child processes:
In process 134599: dagster._core.errors.DagsterInvalidPythonicConfigDefinitionError: 
Error defining Dagster config class <class 'data_pipelines.io_managers.DateIntervalSqlIOManager'> on field 'sql_resource'.
Unable to resolve config type typing.Union[dagster._config.pythonic_config.PartialResource[data_pipelines.resources.SqlResource], data_pipelines.resources.SqlResource] to a supported Dagster config type.
Any ideas? v 1.3.9 Edit: And confirmed on v1.3.11
y

yuhan

06/23/2023, 6:54 PM
mind sharing some code snippets?
b

Brendan Jackson

06/26/2023, 8:41 AM
Sure. Here's my definitions:
Copy code
sql_resource = SqlResource(connection=postgres_connection)
return {
    "dbt": dbt_cli_resource.configured(dbt_config),
    "connection": postgres_connection,
    "sql_resource": sql_resource,
    "sql_io_manager": DateIntervalSqlIOManager(sql_resource=sql_resource),
}
and connection:
Copy code
class ConnectionResource(ConfigurableResource):
    username: str
    password: str
    hostname: str
    port: str
    db_name: str

    def as_string(self) -> str:
        """
        :return: a string representing the database connection
        """
        return get_conn_string(
            username=self.username,
            password=self.password,
            hostname=self.hostname,
            port=self.port,
            db_name=self.db_name,
        )
and
SqlResource
Copy code
class SqlResource(ConfigurableResource):
    connection: ConnectionResource
    _engine: sqlalchemy.engine.Engine = PrivateAttr()

    def setup_for_execution(self, context: InitResourceContext) -> None:
        <http://context.log.info|context.log.info>("Initialising sql alchemy engine")
        self._engine = sqlalchemy.create_engine(self.connection.as_string())

    def get_engine(self):
        return self._engine
This all works fine. But overriding
teardown_after_execution
stops the dependency injection system working.
7 Views