Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[feat] updating metadata #462

Merged
merged 21 commits into from
Sep 28, 2023
Merged

[feat] updating metadata #462

Show file tree
Hide file tree
Changes from 10 commits
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
4ed659b
feat: updating metadata
arthurfg Sep 14, 2023
b6b9efe
feat: removing _atualizado tables
arthurfg Sep 14, 2023
a1e5256
test: testing model registration
arthurfg Sep 14, 2023
feeb7db
fix: fix error message
arthurfg Sep 15, 2023
e000f3a
feat: changing
arthurfg Sep 15, 2023
41abe31
feat: registering more flows
arthurfg Sep 15, 2023
795eefd
fix: fix log
arthurfg Sep 15, 2023
a508426
fx: small fix
arthurfg Sep 15, 2023
9a51294
Merge branch 'master' into staging/including-metadata
arthurfg Sep 22, 2023
ceab71e
[pre-commit.ci] auto fixes from pre-commit.com hooks
pre-commit-ci[bot] Sep 22, 2023
f06efd4
feat: making requested changes
arthurfg Sep 25, 2023
75c5bb1
fix: fixing identation
arthurfg Sep 25, 2023
dd87511
Merge branch 'master' into staging/including-metadata
mergify[bot] Sep 25, 2023
e2ee449
Merge branch 'master' into staging/including-metadata
mergify[bot] Sep 26, 2023
a41181a
Merge branch 'master' into staging/including-metadata
mergify[bot] Sep 27, 2023
2d37128
Merge branch 'master' into staging/including-metadata
arthurfg Sep 27, 2023
0aa3429
[pre-commit.ci] auto fixes from pre-commit.com hooks
pre-commit-ci[bot] Sep 27, 2023
8e57298
fix: changing billing project id
arthurfg Sep 27, 2023
a3016f7
fix: changing billing project id again
arthurfg Sep 27, 2023
e56b915
fix: changing billing project id again again
arthurfg Sep 27, 2023
29b1168
Merge branch 'master' into staging/including-metadata
mergify[bot] Sep 28, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
152 changes: 25 additions & 127 deletions pipelines/datasets/br_anatel_banda_larga_fixa/flows.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
from prefect.storage import GCS
from prefect.tasks.prefect import create_flow_run, wait_for_flow_run
from pipelines.constants import constants
from pipelines.utils.tasks import update_django_metadata
from pipelines.utils.metadata.tasks import update_django_metadata
from pipelines.utils.constants import constants as utils_constants
from pipelines.utils.decorators import Flow
from pipelines.utils.execute_dbt_model.constants import constants as dump_db_constants
Expand Down Expand Up @@ -110,44 +110,18 @@
wait_for_materialization.retry_delay = timedelta(
seconds=dump_db_constants.WAIT_FOR_MATERIALIZATION_RETRY_INTERVAL.value
)

# ! tabela bd pro
with case(materialize_after_dump, True):
# Trigger DBT flow run
current_flow_labels = get_current_flow_labels()
materialization_flow = create_flow_run(
flow_name=utils_constants.FLOW_EXECUTE_DBT_MODEL_NAME.value,
project_name=constants.PREFECT_DEFAULT_PROJECT.value,
parameters={
"dataset_id": dataset_id,
"table_id": table_id[0] + "_atualizado",
"mode": materialization_mode,
"dbt_alias": dbt_alias,
},
labels=current_flow_labels,
run_name=f"Materialize {dataset_id}.{table_id[0]}" + "_atualizado",
)

wait_for_materialization = wait_for_flow_run(
materialization_flow,
stream_states=True,
stream_logs=True,
raise_final_state=True,
)
wait_for_materialization.max_retries = (
dump_db_constants.WAIT_FOR_MATERIALIZATION_RETRY_ATTEMPTS.value
)
wait_for_materialization.retry_delay = timedelta(
seconds=dump_db_constants.WAIT_FOR_MATERIALIZATION_RETRY_INTERVAL.value
)

with case(update_metadata, True):
date = get_today_date_atualizado() # task que retorna a data atual
update_django_metadata(
dataset_id,
table_id[0] + "_atualizado",
table_id[0],
metadata_type="DateTimeRange",
bq_last_update=False,
bq_table_last_year_month=False,
is_bd_pro=True,
is_free=True,
time_delta=2,
time_unit="months",
api_mode="prod",
date_format="yy-mm",
_last_date=date,
Expand Down Expand Up @@ -192,43 +166,18 @@
seconds=dump_db_constants.WAIT_FOR_MATERIALIZATION_RETRY_INTERVAL.value
)

# ! tabela bd pro
with case(materialize_after_dump, True):
# Trigger DBT flow run
current_flow_labels = get_current_flow_labels()
materialization_flow = create_flow_run(
flow_name=utils_constants.FLOW_EXECUTE_DBT_MODEL_NAME.value,
project_name=constants.PREFECT_DEFAULT_PROJECT.value,
parameters={
"dataset_id": dataset_id,
"table_id": table_id[1] + "_atualizado",
"mode": materialization_mode,
"dbt_alias": dbt_alias,
},
labels=current_flow_labels,
run_name=f"Materialize {dataset_id}.{table_id[1]}" + "_atualizado",
)

wait_for_materialization = wait_for_flow_run(
materialization_flow,
stream_states=True,
stream_logs=True,
raise_final_state=True,
)
wait_for_materialization.max_retries = (
dump_db_constants.WAIT_FOR_MATERIALIZATION_RETRY_ATTEMPTS.value
)
wait_for_materialization.retry_delay = timedelta(
seconds=dump_db_constants.WAIT_FOR_MATERIALIZATION_RETRY_INTERVAL.value
)

with case(update_metadata, True):
date = get_today_date_atualizado() # task que retorna a data atual
update_django_metadata(
dataset_id,
table_id[1] + "_atualizado",
table_id[1],
metadata_type="DateTimeRange",
bq_last_update=False,
bq_table_last_year_month=False,
is_bd_pro=True,
is_free=True,
time_delta=2,
time_unit="months",
api_mode="prod",
date_format="yy-mm",
_last_date=date,
Expand Down Expand Up @@ -275,43 +224,18 @@
seconds=dump_db_constants.WAIT_FOR_MATERIALIZATION_RETRY_INTERVAL.value
)

# ! tabela bd pro
with case(materialize_after_dump, True):
# Trigger DBT flow run
current_flow_labels = get_current_flow_labels()
materialization_flow = create_flow_run(
flow_name=utils_constants.FLOW_EXECUTE_DBT_MODEL_NAME.value,
project_name=constants.PREFECT_DEFAULT_PROJECT.value,
parameters={
"dataset_id": dataset_id,
"table_id": table_id[2] + "_atualizado",
"mode": materialization_mode,
"dbt_alias": dbt_alias,
},
labels=current_flow_labels,
run_name=f"Materialize {dataset_id}.{table_id[2]}" + "_atualizado",
)

wait_for_materialization = wait_for_flow_run(
materialization_flow,
stream_states=True,
stream_logs=True,
raise_final_state=True,
)
wait_for_materialization.max_retries = (
dump_db_constants.WAIT_FOR_MATERIALIZATION_RETRY_ATTEMPTS.value
)
wait_for_materialization.retry_delay = timedelta(
seconds=dump_db_constants.WAIT_FOR_MATERIALIZATION_RETRY_INTERVAL.value
)

with case(update_metadata, True):
date = get_today_date_atualizado() # task que retorna a data atual
update_django_metadata(
dataset_id,
table_id[2] + "_atualizado",
table_id[2],
metadata_type="DateTimeRange",
bq_last_update=False,
bq_table_last_year_month=False,
is_bd_pro=True,
is_free=True,
time_delta=2,
time_unit="months",
api_mode="prod",
date_format="yy-mm",
_last_date=date,
Expand Down Expand Up @@ -357,44 +281,18 @@
seconds=dump_db_constants.WAIT_FOR_MATERIALIZATION_RETRY_INTERVAL.value
)

# ! tabela bd pro

with case(materialize_after_dump, True):
# Trigger DBT flow run
current_flow_labels = get_current_flow_labels()
materialization_flow = create_flow_run(
flow_name=utils_constants.FLOW_EXECUTE_DBT_MODEL_NAME.value,
project_name=constants.PREFECT_DEFAULT_PROJECT.value,
parameters={
"dataset_id": dataset_id,
"table_id": table_id[3] + "_atualizado",
"mode": materialization_mode,
"dbt_alias": dbt_alias,
},
labels=current_flow_labels,
run_name=f"Materialize {dataset_id}.{table_id[3]}" + "_atualizado",
)

wait_for_materialization = wait_for_flow_run(
materialization_flow,
stream_states=True,
stream_logs=True,
raise_final_state=True,
)
wait_for_materialization.max_retries = (
dump_db_constants.WAIT_FOR_MATERIALIZATION_RETRY_ATTEMPTS.value
)
wait_for_materialization.retry_delay = timedelta(
seconds=dump_db_constants.WAIT_FOR_MATERIALIZATION_RETRY_INTERVAL.value
)

with case(update_metadata, True):
date = get_today_date_atualizado() # task que retorna a data atual
update_django_metadata(
dataset_id,
table_id[3] + "_atualizado",
table_id[3],
metadata_type="DateTimeRange",
bq_last_update=False,
bq_table_last_year_month=False,
is_bd_pro=True,
is_free=True,
time_delta=2,
time_unit="months",
api_mode="prod",
date_format="yy-mm",
_last_date=date,
Expand Down
Loading
Loading