[Python] How do i pass resources to Jobs so it can be accessible to it's ops

I am new to dagster and I'm trying to pass a resource to a dagster op through the job that it's called from I am having issue even after following the docs, I am not sure if I need to pass config again to jobs nothing seems to work. Here is the code.

Error dagster.core.errors.DagsterInvalidConfigError: Error in config for job Error 1: Missing required config entry "resources" at the root.

import os
from dotenv import load_dotenv
load_dotenv()


@op
def return_one(context):
context.log.info(f'return_one {os.environ.get("BUCKET")}')
return 1


@op(required_resource_keys={"boto3_connection"})
def add_two(context, i: int):
context.log.info(f'##### {context.resources.boto3_connection.get_client()}')
return i + 2


@op
def multi_three(i: int):
return i * 3


class Boto3Connector(object):
def __init__(self, aws_access_key_id, aws_secret_access_key):
self.aws_access_key_id = aws_access_key_id
self.aws_secret_access_key = aws_secret_access_key


def get_client(self, resource="s3"):
session = boto3.session.Session()

session_client = session.client(
service_name=resource,
aws_access_key_id=self.aws_access_key_id,
aws_secret_access_key=self.aws_secret_access_key,
)
return session_client


@resource(
config_schema={
'aws_access_key_id': StringSource,
'aws_secret_access_key': StringSource
})
def boto3_connection(context):
return Boto3Connector(
context.resource_config['aws_access_key_id'],
context.resource_config['aws_secret_access_key']
)


@job(resource_defs={'boto3_connection': boto3_connection})
def my_job():
multi_three(add_two(return_one()))```

Continue reading...