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

Igor Shraga

05/22/2023, 2:17 PM
Hi, Trying to use Pythonic resource. I have this class I want to pass to
@resource
.
Copy code
from dagster import ConfigurableResource


class PostgressConfig(ConfigurableResource):  # type:ignore
    host: str = "my-host"
    port: int = 5432
    database: str = "my-db"
    user: str = "my-user"
    password: str = "my-password"
I try to use it in the following
resource
.
Copy code
@resource(required_resource_keys={"postgress_config"})
def process_repository(context: InitResourceContext) -> ProcessRepository:
    """Resource to provide the process repository to all the ops"""
    if context.log is not None:
        <http://context.log.info|context.log.info>("Creating the Lynceus Platform repository...")
    repository = SqlAlchemyProcessRepository.create(
        host=context.resources.postgress_config.host,
        port=context.resources.postgress_config.port,
        user=context.resources.postgress_config.user,
        password=context.resources.postgress_config.password,
        database=context.resources.postgress_config.database,
    )
    if context.log is not None:
        <http://context.log.info|context.log.info>("Process repository created.")

    return repository
Adding to
definitions
like this.
Copy code
dagster_definitions = Definitions(
    **(PipelineDefinitions.merge([generic_definitions, *project_definitions]).dict()),
    resources={"postgress_config": PostgressConfig()},
)
But I get this error.
Copy code
ImportError while loading conftest '/workspaces/lynceus-platform/test/conftest.py'.
test/conftest.py:51: in <module>
    from pipeline.data.jobs.fill_bi_dashboard import Base as BIDashboardPredictionBase
pipeline/data/jobs/fill_bi_dashboard.py:433: in <module>
    fill_bi_dashboard_job = fill_bi_dashboard_graph.to_job(
/usr/local/lib/python3.10/site-packages/dagster/_core/definitions/graph_definition.py:631: in to_job
    return JobDefinition.dagster_internal_init(
/usr/local/lib/python3.10/site-packages/dagster/_core/definitions/job_definition.py:275: in dagster_internal_init
    return JobDefinition(
/usr/local/lib/python3.10/site-packages/dagster/_core/definitions/job_definition.py:206: in __init__
    self._required_resource_keys = self._get_required_resource_keys(was_provided_resources)
/usr/local/lib/python3.10/site-packages/dagster/_core/definitions/job_definition.py:467: in _get_required_resource_keys
    get_transitive_required_resource_keys(required_keys, self.resource_defs)
/usr/local/lib/python3.10/site-packages/dagster/_core/execution/resources_init.py:388: in get_transitive_required_resource_keys
    ensure_resource_deps_satisfiable(resource_dependencies)
/usr/local/lib/python3.10/site-packages/dagster/_core/execution/resources_init.py:106: in ensure_resource_deps_satisfiable
    _helper(resource_key)
/usr/local/lib/python3.10/site-packages/dagster/_core/execution/resources_init.py:98: in _helper
    raise DagsterInvariantViolationError(
E   dagster._core.errors.DagsterInvariantViolationError: Resource with key 'postgress_config' required by resource with key 'process_repository', but not provided.
It seems that adding
resources
to
defintions
didn't work. Why?
b

ben

05/22/2023, 3:07 PM
Hi Igor, could you share the snippet referred to here? I think this might have to do with trying to construct the job providing some resources to Definitions and some to the job itself
Copy code
pipeline/data/jobs/fill_bi_dashboard.py:433: in <module>
    fill_bi_dashboard_job = fill_bi_dashboard_graph.to_job(
/usr/local/lib/python3.10/site-
i

Igor Shraga

05/22/2023, 3:29 PM
Here:
Copy code
@graph
def fill_bi_dashboard_graph():  # type:ignore
    fill_bi_dashboard_op()


fill_bi_dashboard_job = fill_bi_dashboard_graph.to_job(
    name="fill_bi_dashboard_job",
    resource_defs={
        "process_repository": process_repository,
        "meta_model_repository": meta_model_repository,
        "process_service": process_service,
        "bi_dashboard_repository": bi_dashboard_repository,
    },
    config={
        "ops": {
            "fill_bi_dashboard_op": {
                "config": {
                    "compute_sele
Obviously partial. Hope it's enough.
b

ben

05/22/2023, 6:53 PM
Hi Igor, right now if resources are provided directly to a job (as in the snippet above) Dagster will evaluate to make sure all the resources requirements are satisfied. Since the repository is supplied there, but the postgres config is supplied later, you will see that error. If you supply the postgres config to the job directly,:
Copy code
fill_bi_dashboard_job = fill_bi_dashboard_graph.to_job(
    name="fill_bi_dashboard_job",
    resource_defs={
        "process_repository": process_repository,
        "meta_model_repository": meta_model_repository,
        "process_service": process_service,
        "bi_dashboard_repository": bi_dashboard_repository,
        "postgress_config": PostgressConfig(),
    },
or move the resource defs from the job to the definitions:
Copy code
fill_bi_dashboard_job = fill_bi_dashboard_graph.to_job(
    name="fill_bi_dashboard_job",
    config=...
)

dagster_definitions = Definitions(
    **(PipelineDefinitions.merge([generic_definitions, *project_definitions]).dict()),
    resources={
        "process_repository": process_repository,
        "meta_model_repository": meta_model_repository,
        "process_service": process_service,
        "bi_dashboard_repository": bi_dashboard_repository,
        "postgress_config": PostgressConfig(),
    },
)
either should fix this issue
i

Igor Shraga

05/23/2023, 6:44 AM
The second option sounds great. But, now I have another problem. I added to
Defintions
resources from several different jobs, and now I get
Error 1: Received unexpected config entry "process_repository" at the root. Expected: "{ io_manager?: { config?: Any } }".
. Is there any way around that?
b

ben

05/23/2023, 1:27 PM
Hmm, could you share a bit more from what the
to_job
snippet looks like now? This might be another bit of ordering trickiness
2 Views