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

Nathan Craig

02/10/2023, 5:08 PM
Hello, I am trying to implement oop for a POC since some people at my organization prefer this. However, I am running into an error when using the output of a class method in another function downstream. I tried using the
make_python_type_usable_as_dagster_type
function to convert the pandas.core.frame.DataFrame class to dagster._core.definitions.composition.InvokedNodeOutputHandle but I'm having formatting the inputs correctly. This is the error I get when running the code below without trying to change the type to a dagster type.
dagster._core.errors.DagsterInvalidDefinitionError: In @job ga_fire, received invalid type <class 'pandas.core.frame.DataFrame'> for input "gw_claim" (at position 0) in op invocation "create_final". Must pass the output from previous node invocations or inputs to the composition function as inputs when invoking nodes during composition.
Copy code
#@usable_as_dagster_type
class GW_claim_obj:
    
    #@property
    def __init__(self):
        host = "<http://vpce-0db7ab069278226a9-nmwg1dku.vpce-svc-004d858ae4cced757.us-west-2.vpce.amazonaws.com|vpce-0db7ab069278226a9-nmwg1dku.vpce-svc-004d858ae4cced757.us-west-2.vpce.amazonaws.com>"
        user = os.getenv("REDSHIFT_BDD_LOCAL_UID")
        pw = os.getenv("REDSHIFT_BDD_LOCAL_PWD")
        database = "bi_edw_db"    

        self.con = redshift_connector\
                .connect(
                  host=host,
                  database=database,
                  user=user,
                  password=pw
                  )
    
    #@property
    def create_gw_claim(self):
        #Code for Query Builder in SAS GA Fire Job
        qry = open("gw_claim.sql", "r").read()
        claim_dw_df = pd.read_sql(sql=qry, con=self.con)
        claim_dw_df['quarter'] = claim_dw_df['dateofloss'].dt.to_period('Q').dt.strftime('%Y-%q')
        return claim_dw_df
Here is my job configuration. The other functions are normal functions with @asset decorators.
Copy code
@job
def ga_fire():
    gw_claim_obj = GW_claim_obj()
    gw_claim = gw_claim_obj.create_gw_claim()
    #make_python_type_usable_as_dagster_type(<class 'pandas.core.frame.DataFrame'>,  <class 'dagster._core.definitions.composition.InvokedNodeOutputHandle'>)
    cc_claim = create_cc_claim()
    final = create_final(gw_claim, cc_claim)
    upload_final(final)
Could I get some direction on how to implement class methods in dagster?
s

sean

02/10/2023, 7:18 PM
Hi Nathan, thanks for the question. In general the only thing you should have inside your jobs is op invocations.
usable_as_dagster_type
makes a class usable as a Dagster Type, not as an op. Assets also are not for use inside a
@job
-- it looks to me like you would learn a lot from going through our software-defined asset tutorial: https://docs.dagster.io/tutorial#intro-to-software-defined-assets
n

Nathan Craig

02/13/2023, 4:20 PM
Hi Sean, thanks for the reply. I tried defining all my functions and my class method as ops, but I get an error from dagster not recognizing the self input in a class method shouldn't be required. I might've been asking the wrong question initially, does dagster support using class methods or should we only use function based programming?
Error 1: Missing required config entry "ops" at the root. Sample config for missing entry: {'ops': {'create_gw_claim': {'inputs': {'self': '<selector>'}}}}
s

sean

02/13/2023, 9:18 PM
You can use class methods as you might in any Python program, but not for ops. An op needs to be a plain function (no kind of method).