from dagster import (
define_asset_job,
Definitions,
AssetSelection,
AssetExecutionContext,
DailyPartitionsDefinition
)
from dagster_dbt import DbtCliResource, dbt_assets
from orchestration.settings import DBT_MANIFEST_PATH
from orchestration.utils.dbt import FhirTagAwareDagsterDbtTranslator
@dbt_assets(
manifest=Path(DBT_MANIFEST_PATH),
select="config.materialized:incremental",
dagster_dbt_translator=FhirTagAwareDagsterDbtTranslator(),
)
def partitioned_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
yield from dbt.cli(
["build", "--indirect-selection=buildable"],
context=context,
).stream()
def build_contrib_definitions():
""" Hospital extension point """
# Define a new full refresh job of all incremental assets
full_refresh_incremental_job = define_asset_job(
"chu_arkhn_full_refresh_incrementals",
selection=AssetSelection.assets(
partitioned_dbt_assets
)
)
return {
"jobs": [full_refresh_incremental_job],
"assets": [],
"schedules": [],
}
</details>
<details>
<summary>Lors du merge d'une Pull Request dans la branche **main**, une nouvelle image Docker représentant le livrable est créée, prenant en compte la partie dbt et la partie dagster, qui contient votre extension.</summary>
Vous pouvez dès lors déployer votre code comme d'habitude, et comme décrit dans :
<unknown url="https://www.notion.so/225403a7adf3819a85afc91482fa76be#225403a7adf381758fe9c21eb5169a2f" alt="alias"/>
Enfin, post-redéploiement, vous avez accès à un DAG unifié contenant à la fois vos personnalisations et le code de base de la plateforme data Arkhn :
</details>