Add possibility to use external models as asset dependencies

[lucargir]

Currently when an asset is created, only the SQLMesh dependencies are considered for the Dagster DAG.

dagster-sqlmesh/dagster_sqlmesh/controller/dagster.py

Line 25 in 55d3053

for model, deps in instance.non_external_models_dag():

It would be useful, if the external models could also be added as dependencies to the asset.

This would allow to define SQLMesh models as downstream dependencies of dltHub pipelines for example. The name of the external model could be used to add the dependencies, and ideally, the user could also pass a translator for the external models when defining the SQLMesh asset, so that custom mappings to other Dagster assets can be made.

[ravenac95]

Oh, interesting, we already achieve this for ourselves by the lines here:

dagster-sqlmesh/dagster_sqlmesh/controller/dagster.py

Lines 30 to 33 in 55d3053

	model_deps = [
	SQLMeshModelDep(fqn=dep, model=context.get_model(dep))
	for dep in deps
	]

Is this not working for you at the moment? Though, perhaps we're missing this for sqlmesh's external models? We use the deps to find references to outside models.

[lucargir]

@ravenac95 Actually, I wasn't able to run the project due to dependency conflicts (#31), and was reviewing the code, but it seems I was wrong. Nonetheless, I tried now by cloning the repo and fixing the dependency conflicts, and I see that external dependencies are giving key errors, because for external table paths source.schema.table becomes source_dot__schema_dot__table

So having a translator for external dependencies could come in handy in this case. I tried creating my own translator using SQLMeshDagsterTranslator and replacing the get_asset_key_from_model and get_asset_key_fqn, but then I would get the following error:

internal_deps = {keys_by_output_name[name]: asset_deps[name] for name in asset_deps}
KeyError: source_dot__schema_dot__table

Because it would expect source_dot__schema_dot__table for the name in keys_by_output_name[name]. I set-up the translator to return source__schema__table, but:

dagster-sqlmesh/dagster_sqlmesh/controller/dagster.py

Line 48 in 9d0be7f

model_key = sqlmesh_model_name_to_key(model.name)

is defined by:

dagster-sqlmesh/dagster_sqlmesh/utils.py

Lines 4 to 5 in 9d0be7f

	def sqlmesh_model_name_to_key(name: str) -> str:
	return name.replace(".", "_dot__")

So maybe it should be updated to this?

model_key = translator.get_model_key_fqn(context, dep.fqn)

[ravenac95]

Oh yes I think that's a good change. We should likely have the replacement made by sqlmesh_model_name_to_key be the default.

Would you mind making a PR?

[lucargir]

@ravenac95 Unfortunately, the dependency pinning is an issue, given that fixes that have been issued after the pinned version would not be available, which would impact the project I am working on. So I think I'll develop a temporal fix on my end, and revisit this once SQLMesh reaches v1.0 and has a stable API.

[ravenac95]

@lucargir understood! If I have some time I'll try making some updates to support this use case. I'm also going to update the pinned version but that likely won't help you if you want to continuously update your sqlmesh dependency. I'm honestly not sure that v1.0 will exactly fix the situation still but it should hopefully improve some of the integration. I will likely need to update some things to be a bit more dynamic in terms of the input that is allowed in the interface between sqlmesh + dagster but for now this is not a high priority for our team. The biggest issues is in general the sqlmesh team while awesome has their own dagster integration that is part of the paid product and since we rely on internal mechanisms that are likely never to be stable based on v1 or not v1, this integration, until we determine a better method, might not be so stable without version pinning.

[lucargir]

@ravenac95 just opened PR #41 for this, with some minor improvement on the AssetOut

[lucargir]

PR #41 was merged