[go: nahoru, domu]
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

Jump to bottom

Dataplex operators #20377

Merged
merged 14 commits into from
Mar 14, 2022
Merged

Dataplex operators #20377

Changes from 1 commit
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
9d47be5
Add dataplex operators
Dec 10, 2021
637e783
Add impersonation_chain support. Implement sensor tests
Dec 17, 2021
ae7d274
Fix unused imports
Dec 20, 2021
d385ee2
Update logging. Add type info in hook & TYPE_CHECKING
Jan 5, 2022
2d0c04e
Remove API_KEY
Jan 12, 2022
f2850dc
Rewrite from raw API to Python client
Mar 1, 2022
d53678f
Add links
Mar 2, 2022
7747a7e
Update links
Mar 4, 2022
8aae7c6
Update unit tests
Mar 7, 2022
b5d0f2b
Remove types. Update system tests
Mar 8, 2022
da2e7fa
Fix static checks
Mar 9, 2022
06d5186
Remove constant timeout. Fix metadata & timeout type
Mar 9, 2022
36e0092
Revert changed setup
Mar 9, 2022
96a62fd
Fix sensor test
Mar 10, 2022
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
Prev Previous commit
Next Next commit
Update links
  • Loading branch information
Wojciech Januszek committed Mar 14, 2022
commit 7747a7e237ad62749faf46798cdaac1f0dda76ff
44 changes: 5 additions & 39 deletions airflow/providers/google/cloud/links/dataplex.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,10 +17,9 @@
# under the License.
"""This module contains Google Dataplex links."""

from datetime import datetime
from typing import TYPE_CHECKING

from airflow.models import BaseOperator, BaseOperatorLink, XCom
from airflow.providers.google.cloud.links.base import BaseGoogleLink

if TYPE_CHECKING:
from airflow.utils.context import Context
Expand All @@ -30,11 +29,12 @@
DATAPLEX_TASKS_LINK = DATAPLEX_BASE_LINK + "?project={project_id}&qLake={lake_id}.{region}"


class DataplexTaskLink(BaseOperatorLink):
class DataplexTaskLink(BaseGoogleLink):
"""Helper class for constructing Dataplex Task link"""

name = "Dataplex Task"
key = "task_conf"
format_str = DATAPLEX_TASK_LINK

@staticmethod
def persist(
Expand All @@ -52,30 +52,13 @@ def persist(
},
)

def get_link(self, operator: BaseOperator, dttm: datetime):
task_conf = XCom.get_one(
key=DataplexTaskLink.key,
dag_id=operator.dag.dag_id,
task_id=operator.task_id,
execution_date=dttm,
)
return (
DATAPLEX_TASK_LINK.format(
lake_id=task_conf["lake_id"],
task_id=task_conf["task_id"],
region=task_conf["region"],
project_id=task_conf["project_id"],
)
if task_conf
else ""
)


class DataplexTasksLink(BaseOperatorLink):
class DataplexTasksLink(BaseGoogleLink):
"""Helper class for constructing Dataplex Tasks link"""

name = "Dataplex Tasks"
key = "tasks_conf"
format_str = DATAPLEX_TASKS_LINK

@staticmethod
def persist(
Expand All @@ -91,20 +74,3 @@ def persist(
"region": task_instance.region,
},
)

def get_link(self, operator: BaseOperator, dttm: datetime):
tasks_conf = XCom.get_one(
key=DataplexTasksLink.key,
dag_id=operator.dag.dag_id,
task_id=operator.task_id,
execution_date=dttm,
)
return (
DATAPLEX_TASKS_LINK.format(
project_id=tasks_conf["project_id"],
lake_id=tasks_conf["lake_id"],
region=tasks_conf["region"],
)
if tasks_conf
else ""
)