This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new d1aa509bbd D205 Support - Providers: Databricks to Github (inclusive)
(#32243)
d1aa509bbd is described below
commit d1aa509bbd1941ceb3fe31789efeebbddd58d32f
Author: D. Ferruzzi <[email protected]>
AuthorDate: Wed Jun 28 11:23:34 2023 -0700
D205 Support - Providers: Databricks to Github (inclusive) (#32243)
---
airflow/providers/databricks/hooks/databricks.py | 3 +-
.../providers/databricks/hooks/databricks_base.py | 1 +
.../providers/databricks/operators/databricks.py | 14 ++++-----
.../databricks/operators/databricks_repos.py | 19 ++++++------
.../databricks/operators/databricks_sql.py | 1 +
.../databricks/sensors/databricks_partition.py | 9 ++----
airflow/providers/databricks/utils/databricks.py | 16 +++++-----
airflow/providers/datadog/hooks/datadog.py | 14 ++++-----
airflow/providers/datadog/sensors/datadog.py | 6 ++--
airflow/providers/dbt/cloud/hooks/dbt.py | 36 +++++++++++++---------
airflow/providers/dbt/cloud/operators/dbt.py | 9 ++----
airflow/providers/dingding/operators/dingding.py | 3 +-
airflow/providers/discord/hooks/discord_webhook.py | 7 ++---
.../providers/discord/operators/discord_webhook.py | 1 +
airflow/providers/docker/decorators/docker.py | 3 +-
airflow/providers/docker/operators/docker_swarm.py | 1 +
.../providers/elasticsearch/hooks/elasticsearch.py | 1 +
.../elasticsearch/log/es_json_formatter.py | 5 +--
.../providers/elasticsearch/log/es_task_handler.py | 27 ++++++++--------
airflow/providers/ftp/hooks/ftp.py | 8 ++---
airflow/providers/ftp/operators/ftp.py | 8 ++---
airflow/providers/github/sensors/github.py | 5 +--
22 files changed, 94 insertions(+), 103 deletions(-)
diff --git a/airflow/providers/databricks/hooks/databricks.py
b/airflow/providers/databricks/hooks/databricks.py
index 4d1aced4e1..e880a42eee 100644
--- a/airflow/providers/databricks/hooks/databricks.py
+++ b/airflow/providers/databricks/hooks/databricks.py
@@ -459,7 +459,8 @@ class DatabricksHook(BaseDatabricksHook):
def get_repo_by_path(self, path: str) -> str | None:
"""
- Obtains Repos ID by path
+ Obtains Repos ID by path.
+
:param path: path to a repository
:return: Repos ID if it exists, None if doesn't.
"""
diff --git a/airflow/providers/databricks/hooks/databricks_base.py
b/airflow/providers/databricks/hooks/databricks_base.py
index cf48fc4b00..9885e9a998 100644
--- a/airflow/providers/databricks/hooks/databricks_base.py
+++ b/airflow/providers/databricks/hooks/databricks_base.py
@@ -543,6 +543,7 @@ class BaseDatabricksHook(BaseHook):
async def _a_do_api_call(self, endpoint_info: tuple[str, str], json:
dict[str, Any] | None = None):
"""
Async version of `_do_api_call()`.
+
:param endpoint_info: Tuple of method and endpoint
:param json: Parameters for this API call.
:return: If the api call returns a OK status code,
diff --git a/airflow/providers/databricks/operators/databricks.py
b/airflow/providers/databricks/operators/databricks.py
index 305e410a01..ab93d8f49b 100644
--- a/airflow/providers/databricks/operators/databricks.py
+++ b/airflow/providers/databricks/operators/databricks.py
@@ -162,10 +162,9 @@ class DatabricksJobRunLink(BaseOperatorLink):
class DatabricksSubmitRunOperator(BaseOperator):
"""
- Submits a Spark job run to Databricks using the
- `api/2.1/jobs/runs/submit
-
<https://docs.databricks.com/dev-tools/api/latest/jobs.html#operation/JobsRunsSubmit>`_
- API endpoint.
+ Submits a Spark job run to Databricks using the api/2.1/jobs/runs/submit
API endpoint.
+
+ See:
https://docs.databricks.com/dev-tools/api/latest/jobs.html#operation/JobsRunsSubmit
There are three ways to instantiate this operator.
@@ -428,10 +427,9 @@ class
DatabricksSubmitRunDeferrableOperator(DatabricksSubmitRunOperator):
class DatabricksRunNowOperator(BaseOperator):
"""
- Runs an existing Spark job run to Databricks using the
- `api/2.1/jobs/run-now
-
<https://docs.databricks.com/dev-tools/api/latest/jobs.html#operation/JobsRunNow>`_
- API endpoint.
+ Runs an existing Spark job run to Databricks using the
api/2.1/jobs/run-now API endpoint.
+
+ See:
https://docs.databricks.com/dev-tools/api/latest/jobs.html#operation/JobsRunNow
There are two ways to instantiate this operator.
diff --git a/airflow/providers/databricks/operators/databricks_repos.py
b/airflow/providers/databricks/operators/databricks_repos.py
index 5e1bf69964..64735cadbf 100644
--- a/airflow/providers/databricks/operators/databricks_repos.py
+++ b/airflow/providers/databricks/operators/databricks_repos.py
@@ -33,10 +33,9 @@ if TYPE_CHECKING:
class DatabricksReposCreateOperator(BaseOperator):
"""
- Creates a Databricks Repo
- using
- `POST api/2.0/repos
<https://docs.databricks.com/dev-tools/api/latest/repos.html#operation/create-repo>`_
- API endpoint and optionally checking it out to a specific branch or tag.
+ Creates, and optionally checks out, a Databricks Repo using the POST
api/2.0/repos API endpoint.
+
+ See:
https://docs.databricks.com/dev-tools/api/latest/repos.html#operation/create-repo
:param git_url: Required HTTPS URL of a Git repository
:param git_provider: Optional name of Git provider. Must be provided if we
can't guess its name from URL.
@@ -165,9 +164,9 @@ class DatabricksReposCreateOperator(BaseOperator):
class DatabricksReposUpdateOperator(BaseOperator):
"""
- Updates specified repository to a given branch or tag
- using `PATCH api/2.0/repos
-
<https://docs.databricks.com/dev-tools/api/latest/repos.html#operation/update-repo>`_
API endpoint.
+ Updates specified repository to a given branch or tag using the PATCH
api/2.0/repos API endpoint.
+
+ See:
https://docs.databricks.com/dev-tools/api/latest/repos.html#operation/update-repo
:param branch: optional name of branch to update to. Should be specified
if ``tag`` is omitted
:param tag: optional name of tag to update to. Should be specified if
``branch`` is omitted
@@ -241,9 +240,9 @@ class DatabricksReposUpdateOperator(BaseOperator):
class DatabricksReposDeleteOperator(BaseOperator):
"""
- Deletes specified repository
- using `DELETE api/2.0/repos
-
<https://docs.databricks.com/dev-tools/api/latest/repos.html#operation/delete-repo>`_
API endpoint.
+ Deletes specified repository using the DELETE api/2.0/repos API endpoint.
+
+ See:
https://docs.databricks.com/dev-tools/api/latest/repos.html#operation/delete-repo
:param repo_id: optional ID of existing repository. Should be specified if
``repo_path`` is omitted
:param repo_path: optional path of existing repository. Should be
specified if ``repo_id`` is omitted
diff --git a/airflow/providers/databricks/operators/databricks_sql.py
b/airflow/providers/databricks/operators/databricks_sql.py
index ea2f1d0915..663b4ef0fe 100644
--- a/airflow/providers/databricks/operators/databricks_sql.py
+++ b/airflow/providers/databricks/operators/databricks_sql.py
@@ -172,6 +172,7 @@ COPY_INTO_APPROVED_FORMATS = ["CSV", "JSON", "AVRO", "ORC",
"PARQUET", "TEXT", "
class DatabricksCopyIntoOperator(BaseOperator):
"""
Executes COPY INTO command in a Databricks SQL endpoint or a Databricks
cluster.
+
COPY INTO command is constructed from individual pieces, that are
described in
`documentation
<https://docs.databricks.com/sql/language-manual/delta-copy-into.html>`_.
diff --git a/airflow/providers/databricks/sensors/databricks_partition.py
b/airflow/providers/databricks/sensors/databricks_partition.py
index b848415292..fe50c6e1b3 100644
--- a/airflow/providers/databricks/sensors/databricks_partition.py
+++ b/airflow/providers/databricks/sensors/databricks_partition.py
@@ -136,13 +136,7 @@ class DatabricksPartitionSensor(BaseSensorOperator):
)
def _check_table_partitions(self) -> list:
- """
- The method performs the following:
- * Generates the fully qualified table name.
- * Calls the generate partition query.
- * Based on the result returned by the partition generation method,
- the _sql_sensor method is called.
- """
+ """Generate the fully qualified table name, generate partition, and
call the _sql_sensor method."""
if self.table_name.split(".")[0] == "delta":
_fully_qualified_table_name = self.table_name
else:
@@ -173,6 +167,7 @@ class DatabricksPartitionSensor(BaseSensorOperator):
) -> str:
"""
Queries the table for available partitions.
+
Generates the SQL query based on the partition data types.
* For a list, it prepares the SQL in the format:
column_name in (value1, value2,...)
diff --git a/airflow/providers/databricks/utils/databricks.py
b/airflow/providers/databricks/utils/databricks.py
index 0f32d647f0..050eba9096 100644
--- a/airflow/providers/databricks/utils/databricks.py
+++ b/airflow/providers/databricks/utils/databricks.py
@@ -23,11 +23,12 @@ from airflow.providers.databricks.hooks.databricks import
RunState
def normalise_json_content(content, json_path: str = "json") -> str | bool |
list | dict:
"""
- Normalize content or all values of content if it is a dict to a string. The
- function will throw if content contains non-string or non-numeric
non-boolean types.
- The reason why we have this function is because the ``self.json`` field
must be a
- dict with only string values. This is because ``render_template`` will fail
- for numerical values.
+ Normalize content or all values of content if it is a dict to a string.
+
+ The function will throw if content contains non-string or non-numeric
non-boolean
+ types. The reason why we have this function is because the ``self.json``
field
+ must be a dict with only string values. This is because
``render_template`` will
+ fail for numerical values.
The only one exception is when we have boolean values, they can not be
converted
to string type because databricks does not understand 'True' or 'False'
values.
@@ -56,8 +57,9 @@ def normalise_json_content(content, json_path: str = "json")
-> str | bool | lis
def validate_trigger_event(event: dict):
"""
- Validates correctness of the event
- received from
:class:`~airflow.providers.databricks.triggers.databricks.DatabricksExecutionTrigger`.
+ Validates correctness of the event received from
DatabricksExecutionTrigger.
+
+ See:
:class:`~airflow.providers.databricks.triggers.databricks.DatabricksExecutionTrigger`.
"""
keys_to_check = ["run_id", "run_page_url", "run_state"]
for key in keys_to_check:
diff --git a/airflow/providers/datadog/hooks/datadog.py
b/airflow/providers/datadog/hooks/datadog.py
index 899de6e4c9..eae101eabd 100644
--- a/airflow/providers/datadog/hooks/datadog.py
+++ b/airflow/providers/datadog/hooks/datadog.py
@@ -29,8 +29,9 @@ from airflow.utils.log.logging_mixin import LoggingMixin
class DatadogHook(BaseHook, LoggingMixin):
"""
- Uses datadog API to send metrics of practically anything measurable,
- so it's possible to track # of db records inserted/deleted, records read
+ Uses datadog API to send metrics of practically anything measurable.
+
+ It's possible to track # of db records inserted/deleted, records read
from file and many other useful metrics.
Depends on the datadog API, which has to be deployed on the same server
where
@@ -94,8 +95,7 @@ class DatadogHook(BaseHook, LoggingMixin):
def query_metric(self, query: str, from_seconds_ago: int, to_seconds_ago:
int) -> dict[str, Any]:
"""
- Queries datadog for a specific metric, potentially with some
- function applied to it and returns the results.
+ Query datadog for a metric, potentially with some function applied to
it and return the result.
:param query: The datadog query to execute (see datadog docs)
:param from_seconds_ago: How many seconds ago to start querying for.
@@ -122,9 +122,9 @@ class DatadogHook(BaseHook, LoggingMixin):
device_name: list[str] | None = None,
) -> dict[str, Any]:
"""
- Posts an event to datadog (processing finished, potentially alerts,
other issues)
- Think about this as a means to maintain persistence of alerts, rather
than
- alerting itself.
+ Posts an event to datadog (processing finished, potentially alerts,
other issues).
+
+ Think about this as a means to maintain persistence of alerts, rather
than alerting itself.
:param title: The title of the event
:param text: The body of the event (more information)
diff --git a/airflow/providers/datadog/sensors/datadog.py
b/airflow/providers/datadog/sensors/datadog.py
index a94f6ccc3c..da80049e66 100644
--- a/airflow/providers/datadog/sensors/datadog.py
+++ b/airflow/providers/datadog/sensors/datadog.py
@@ -31,11 +31,9 @@ if TYPE_CHECKING:
class DatadogSensor(BaseSensorOperator):
"""
- A sensor to listen, with a filter, to datadog event streams and determine
- if some event was emitted.
+ A sensor to listen, with a filter, to datadog event streams and determine
if some event was emitted.
- Depends on the datadog API, which has to be deployed on the same server
where
- Airflow runs.
+ Depends on the datadog API, which has to be deployed on the same server
where Airflow runs.
:param datadog_conn_id: The connection to datadog, containing metadata for
api keys.
:param from_seconds_ago: POSIX timestamp start (default 3600).
diff --git a/airflow/providers/dbt/cloud/hooks/dbt.py
b/airflow/providers/dbt/cloud/hooks/dbt.py
index 4d57103689..fdbffc4b34 100644
--- a/airflow/providers/dbt/cloud/hooks/dbt.py
+++ b/airflow/providers/dbt/cloud/hooks/dbt.py
@@ -41,8 +41,10 @@ if TYPE_CHECKING:
def fallback_to_default_account(func: Callable) -> Callable:
"""
- Decorator which provides a fallback value for ``account_id``. If the
``account_id`` is None or not passed
- to the decorated function, the value will be taken from the configured dbt
Cloud Airflow Connection.
+ Decorator which provides a fallback value for ``account_id``.
+
+ If the ``account_id`` is None or not passed to the decorated function,
+ the value will be taken from the configured dbt Cloud Airflow Connection.
"""
sig = signature(func)
@@ -134,8 +136,10 @@ T = TypeVar("T", bound=Any)
def provide_account_id(func: T) -> T:
"""
- Decorator which provides a fallback value for ``account_id``. If the
``account_id`` is None or not passed
- to the decorated function, the value will be taken from the configured dbt
Cloud Airflow Connection.
+ Decorator which provides a fallback value for ``account_id``.
+
+ If the ``account_id`` is None or not passed to the decorated function,
+ the value will be taken from the configured dbt Cloud Airflow Connection.
"""
function_signature = signature(func)
@@ -362,8 +366,9 @@ class DbtCloudHook(HttpHook):
project_id: int | None = None,
) -> list[Response]:
"""
- Retrieves metadata for all jobs tied to a specified dbt Cloud account.
If a ``project_id`` is
- supplied, only jobs pertaining to this project will be retrieved.
+ Retrieves metadata for all jobs tied to a specified dbt Cloud account.
+
+ If a ``project_id`` is supplied, only jobs pertaining to this project
will be retrieved.
:param account_id: Optional. The ID of a dbt Cloud account.
:param order_by: Optional. Field to order the result by. Use '-' to
indicate reverse order.
@@ -437,8 +442,9 @@ class DbtCloudHook(HttpHook):
order_by: str | None = None,
) -> list[Response]:
"""
- Retrieves metadata for all of the dbt Cloud job runs for an account.
If a ``job_definition_id`` is
- supplied, only metadata for runs of that specific job are pulled.
+ Retrieves metadata for all dbt Cloud job runs for an account.
+
+ If a ``job_definition_id`` is supplied, only metadata for runs of that
specific job are pulled.
:param account_id: Optional. The ID of a dbt Cloud account.
:param include_related: Optional. List of related fields to pull with
the run.
@@ -555,9 +561,10 @@ class DbtCloudHook(HttpHook):
self, run_id: int, account_id: int | None = None, step: int | None =
None
) -> list[Response]:
"""
- Retrieves a list of the available artifact files generated for a
completed run of a dbt Cloud job. By
- default, this returns artifacts from the last step in the run. To list
artifacts from other steps in
- the run, use the ``step`` parameter.
+ Retrieves a list of the available artifact files generated for a
completed run of a dbt Cloud job.
+
+ By default, this returns artifacts from the last step in the run. To
+ list artifacts from other steps in the run, use the ``step`` parameter.
:param run_id: The ID of a dbt Cloud job run.
:param account_id: Optional. The ID of a dbt Cloud account.
@@ -575,9 +582,10 @@ class DbtCloudHook(HttpHook):
self, run_id: int, path: str, account_id: int | None = None, step: int
| None = None
) -> Response:
"""
- Retrieves a list of the available artifact files generated for a
completed run of a dbt Cloud job. By
- default, this returns artifacts from the last step in the run. To list
artifacts from other steps in
- the run, use the ``step`` parameter.
+ Retrieves a list of the available artifact files generated for a
completed run of a dbt Cloud job.
+
+ By default, this returns artifacts from the last step in the run. To
+ list artifacts from other steps in the run, use the ``step`` parameter.
:param run_id: The ID of a dbt Cloud job run.
:param path: The file path related to the artifact file. Paths are
rooted at the target/ directory.
diff --git a/airflow/providers/dbt/cloud/operators/dbt.py
b/airflow/providers/dbt/cloud/operators/dbt.py
index d9177e6772..f316c47f3d 100644
--- a/airflow/providers/dbt/cloud/operators/dbt.py
+++ b/airflow/providers/dbt/cloud/operators/dbt.py
@@ -37,10 +37,7 @@ if TYPE_CHECKING:
class DbtCloudRunJobOperatorLink(BaseOperatorLink):
- """
- Operator link for DbtCloudRunJobOperator. This link allows users to
monitor the triggered job run
- directly in dbt Cloud.
- """
+ """Allows users to monitor the triggered job run directly in dbt Cloud."""
name = "Monitor Job Run"
@@ -192,8 +189,8 @@ class DbtCloudRunJobOperator(BaseOperator):
def execute_complete(self, context: Context, event: dict[str, Any]) -> int:
"""
Callback for when the trigger fires - returns immediately.
- Relies on trigger to throw an exception, otherwise it assumes
execution was
- successful.
+
+ Relies on trigger to throw an exception, otherwise it assumes
execution was successful.
"""
if event["status"] == "error":
raise AirflowException(event["message"])
diff --git a/airflow/providers/dingding/operators/dingding.py
b/airflow/providers/dingding/operators/dingding.py
index 6152ddefa6..49db1a6870 100644
--- a/airflow/providers/dingding/operators/dingding.py
+++ b/airflow/providers/dingding/operators/dingding.py
@@ -28,7 +28,8 @@ if TYPE_CHECKING:
class DingdingOperator(BaseOperator):
"""
- This operator allows you send Dingding message using Dingding custom bot.
+ This operator allows you to send Dingding message using Dingding custom
bot.
+
Get Dingding token from conn_id.password. And prefer set domain to
conn_id.host, if not will use default ``https://oapi.dingtalk.com``.
diff --git a/airflow/providers/discord/hooks/discord_webhook.py
b/airflow/providers/discord/hooks/discord_webhook.py
index 2bfdebb18a..76fb9ac52f 100644
--- a/airflow/providers/discord/hooks/discord_webhook.py
+++ b/airflow/providers/discord/hooks/discord_webhook.py
@@ -28,6 +28,7 @@ from airflow.providers.http.hooks.http import HttpHook
class DiscordWebhookHook(HttpHook):
"""
This hook allows you to post messages to Discord using incoming webhooks.
+
Takes a Discord connection ID with a default relative webhook endpoint. The
default endpoint can be overridden using the webhook_endpoint parameter
(https://discordapp.com/developers/docs/resources/webhook).
@@ -76,8 +77,7 @@ class DiscordWebhookHook(HttpHook):
def _get_webhook_endpoint(self, http_conn_id: str | None,
webhook_endpoint: str | None) -> str:
"""
- Given a Discord http_conn_id, return the default webhook endpoint or
override if a
- webhook_endpoint is manually supplied.
+ Return the default webhook endpoint or override if a webhook_endpoint
is manually supplied.
:param http_conn_id: The provided connection ID
:param webhook_endpoint: The manually provided webhook endpoint
@@ -104,8 +104,7 @@ class DiscordWebhookHook(HttpHook):
def _build_discord_payload(self) -> str:
"""
- Construct the Discord JSON payload. All relevant parameters are
combined here
- to a valid Discord JSON payload.
+ Combine all relevant parameters into a valid Discord JSON payload.
:return: Discord payload (str) to send
"""
diff --git a/airflow/providers/discord/operators/discord_webhook.py
b/airflow/providers/discord/operators/discord_webhook.py
index d72d3a8298..e4c38f8356 100644
--- a/airflow/providers/discord/operators/discord_webhook.py
+++ b/airflow/providers/discord/operators/discord_webhook.py
@@ -30,6 +30,7 @@ if TYPE_CHECKING:
class DiscordWebhookOperator(SimpleHttpOperator):
"""
This operator allows you to post messages to Discord using incoming
webhooks.
+
Takes a Discord connection ID with a default relative webhook endpoint. The
default endpoint can be overridden using the webhook_endpoint parameter
(https://discordapp.com/developers/docs/resources/webhook).
diff --git a/airflow/providers/docker/decorators/docker.py
b/airflow/providers/docker/decorators/docker.py
index 1ea76f248d..bb7849f1b2 100644
--- a/airflow/providers/docker/decorators/docker.py
+++ b/airflow/providers/docker/decorators/docker.py
@@ -150,7 +150,8 @@ def docker_task(
**kwargs,
) -> TaskDecorator:
"""
- Python operator decorator. Wraps a function into an Airflow operator.
+ Python operator decorator; wraps a function into an Airflow operator.
+
Also accepts any argument that DockerOperator will via ``kwargs``. Can be
reused in a single DAG.
:param python_callable: Function to decorate
diff --git a/airflow/providers/docker/operators/docker_swarm.py
b/airflow/providers/docker/operators/docker_swarm.py
index 88377871c3..9b68b92aa1 100644
--- a/airflow/providers/docker/operators/docker_swarm.py
+++ b/airflow/providers/docker/operators/docker_swarm.py
@@ -32,6 +32,7 @@ if TYPE_CHECKING:
class DockerSwarmOperator(DockerOperator):
"""
Execute a command as an ephemeral docker swarm service.
+
Example use-case - Using Docker Swarm orchestration to make one-time
scripts highly available.
diff --git a/airflow/providers/elasticsearch/hooks/elasticsearch.py
b/airflow/providers/elasticsearch/hooks/elasticsearch.py
index bc55762a15..978eac8f5c 100644
--- a/airflow/providers/elasticsearch/hooks/elasticsearch.py
+++ b/airflow/providers/elasticsearch/hooks/elasticsearch.py
@@ -141,6 +141,7 @@ class ElasticsearchSQLHook(DbApiHook):
class ElasticsearchHook(ElasticsearchSQLHook):
"""
This class is deprecated and was renamed to ElasticsearchSQLHook.
+
Please use
`airflow.providers.elasticsearch.hooks.elasticsearch.ElasticsearchSQLHook`.
"""
diff --git a/airflow/providers/elasticsearch/log/es_json_formatter.py
b/airflow/providers/elasticsearch/log/es_json_formatter.py
index 0cc204c208..7ac543c0a3 100644
--- a/airflow/providers/elasticsearch/log/es_json_formatter.py
+++ b/airflow/providers/elasticsearch/log/es_json_formatter.py
@@ -29,10 +29,7 @@ class ElasticsearchJSONFormatter(JSONFormatter):
default_tz_format = "%z"
def formatTime(self, record, datefmt=None):
- """
- Returns the creation time of the specified LogRecord in ISO 8601 date
and time format
- in the local time zone.
- """
+ """Return the creation time of the LogRecord in ISO 8601 date/time
format in the local time zone."""
dt = pendulum.from_timestamp(record.created,
tz=pendulum.local_timezone())
if datefmt:
s = dt.strftime(datefmt)
diff --git a/airflow/providers/elasticsearch/log/es_task_handler.py
b/airflow/providers/elasticsearch/log/es_task_handler.py
index 8c8847fe1a..1bc071d2c2 100644
--- a/airflow/providers/elasticsearch/log/es_task_handler.py
+++ b/airflow/providers/elasticsearch/log/es_task_handler.py
@@ -82,12 +82,12 @@ class ElasticSearchResponse:
class ElasticsearchTaskHandler(FileTaskHandler, ExternalLoggingMixin,
LoggingMixin):
"""
- ElasticsearchTaskHandler is a python log handler that
- reads logs from Elasticsearch. Note that Airflow does not handle the
indexing
- of logs into Elasticsearch. Instead, Airflow flushes logs
- into local files. Additional software setup is required
- to index the logs into Elasticsearch, such as using
- Filebeat and Logstash.
+ ElasticsearchTaskHandler is a python log handler that reads logs from
Elasticsearch.
+
+ Note that Airflow does not handle the indexing of logs into Elasticsearch.
Instead,
+ Airflow flushes logs into local files. Additional software setup is
required to index
+ the logs into Elasticsearch, such as using Filebeat and Logstash.
+
To efficiently query and sort Elasticsearch results, this handler assumes
each
log message has a field `log_id` consists of ti primary keys:
`log_id = {dag_id}-{task_id}-{execution_date}-{try_number}`
@@ -95,6 +95,10 @@ class ElasticsearchTaskHandler(FileTaskHandler,
ExternalLoggingMixin, LoggingMix
which is a unique integer indicates log message's order.
Timestamps here are unreliable because multiple log messages
might have the same timestamp.
+
+ :param base_log_folder: base folder to store logs locally
+ :param log_id_template: log id template
+ :param host: Elasticsearch host name
"""
PAGE = 0
@@ -120,11 +124,6 @@ class ElasticsearchTaskHandler(FileTaskHandler,
ExternalLoggingMixin, LoggingMix
filename_template: str | None = None,
log_id_template: str | None = None,
):
- """
- :param base_log_folder: base folder to store logs locally
- :param log_id_template: log id template
- :param host: Elasticsearch host name
- """
es_kwargs = es_kwargs or {}
super().__init__(base_log_folder, filename_template)
self.closed = False
@@ -198,8 +197,7 @@ class ElasticsearchTaskHandler(FileTaskHandler,
ExternalLoggingMixin, LoggingMix
@staticmethod
def _clean_date(value: datetime | None) -> str:
"""
- Clean up a date value so that it is safe to query in elasticsearch
- by removing reserved characters.
+ Clean up a date value so that it is safe to query in elasticsearch by
removing reserved characters.
https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl-query-string-query.html#_reserved_characters
"""
@@ -303,8 +301,7 @@ class ElasticsearchTaskHandler(FileTaskHandler,
ExternalLoggingMixin, LoggingMix
def es_read(self, log_id: str, offset: str, metadata: dict) -> list:
"""
- Returns the logs matching log_id in Elasticsearch and next offset.
- Returns '' if no log is found or there was an error.
+ Return the logs matching log_id in Elasticsearch and next offset or ''.
:param log_id: the log_id of the log to read.
:param offset: the offset start to read log from.
diff --git a/airflow/providers/ftp/hooks/ftp.py
b/airflow/providers/ftp/hooks/ftp.py
index c0ff2733ce..26080aaa69 100644
--- a/airflow/providers/ftp/hooks/ftp.py
+++ b/airflow/providers/ftp/hooks/ftp.py
@@ -65,18 +65,14 @@ class FTPHook(BaseHook):
return self.conn
def close_conn(self):
- """
- Closes the connection. An error will occur if the
- connection was not ever opened.
- """
+ """Closes the connection; an error will occur if the connection was
never opened."""
conn = self.conn
conn.quit()
self.conn = None
def describe_directory(self, path: str) -> dict:
"""
- Returns a dictionary of {filename: {attributes}} for all files
- on the remote system (where the MLSD command is supported).
+ Return a dictionary of {filename: {attributes}} for all files on a
remote system which supports MLSD.
:param path: full path to the remote directory
"""
diff --git a/airflow/providers/ftp/operators/ftp.py
b/airflow/providers/ftp/operators/ftp.py
index d8a0246495..8c935751d4 100644
--- a/airflow/providers/ftp/operators/ftp.py
+++ b/airflow/providers/ftp/operators/ftp.py
@@ -37,8 +37,8 @@ class FTPOperation:
class FTPFileTransmitOperator(BaseOperator):
"""
FTPFileTransmitOperator for transferring files from remote host to local
or vice a versa.
- This operator uses an FTPHook to open ftp transport channel that serve as
basis
- for file transfer.
+
+ This operator uses an FTPHook to open ftp transport channel that serve as
basis for file transfer.
.. seealso::
For more information on how to use this operator, take a look at the
guide:
@@ -139,8 +139,8 @@ class FTPFileTransmitOperator(BaseOperator):
class FTPSFileTransmitOperator(FTPFileTransmitOperator):
"""
FTPSFileTransmitOperator for transferring files from remote host to local
or vice a versa.
- This operator uses an FTPSHook to open ftps transport channel that serve
as basis
- for file transfer.
+
+ This operator uses an FTPSHook to open ftps transport channel that serve
as basis for file transfer.
.. seealso::
For more information on how to use this operator, take a look at the
guide:
diff --git a/airflow/providers/github/sensors/github.py
b/airflow/providers/github/sensors/github.py
index c948635d41..eb9caad5ae 100644
--- a/airflow/providers/github/sensors/github.py
+++ b/airflow/providers/github/sensors/github.py
@@ -91,10 +91,7 @@ class BaseGithubRepositorySensor(GithubSensor):
)
def poke(self, context: Context) -> bool:
- """
- Function that the sensors defined while deriving this class should
- override.
- """
+ """Function that the sensors defined while deriving this class should
override."""
raise AirflowException("Override me.")
