[dagster-pipes] add centralized dict for common Dagster tags

Author: danielgafni

python_modules/dagster/dagster/_core/pipes/context.py

@@ -1,7 +1,8 @@
+import os
 from contextlib import contextmanager
 from dataclasses import dataclass
 from queue import Queue
-from typing import TYPE_CHECKING, Any, Iterator, Mapping, Optional, Sequence, Union, cast
+from typing import TYPE_CHECKING, Any, Dict, Iterator, Mapping, Optional, Sequence, Union, cast
 
 from dagster_pipes import (
     DAGSTER_PIPES_CONTEXT_ENV_VAR,
@@ -275,6 +276,40 @@ class PipesSession:
     message_reader_params: PipesParams
     context: OpExecutionContext
 
+    @property
+    def dagster_tags(self) -> Dict[str, str]:
+        """Important Dagster key-value pairs typically attached to external resources launched from Pipes."""
+        tags = {
+            "dagster/run-id": self.context.run_id,
+            "dagster/job": self.context.job_name,
+        }
+
+        if self.context.dagster_run.external_job_origin:
+            tags["dagster/code-location"] = (
+                self.context.dagster_run.external_job_origin.repository_origin.code_location_origin.location_name
+            )
+
+        if user := self.context.get_tag("dagster/user"):
+            tags["dagster/user"] = user
+
+        if self.context.has_partition_key:
+            tags["dagster/partition-key"] = self.context.partition_key
+
+        # now using the walrus operator for os.getenv("DAGSTER_CLOUD_DEPLOYMENT_NAME")
+
+        for dagster_cloud_env_var in [
+            "DAGSTER_CLOUD_DEPLOYMENT_NAME",
+            "DAGSTER_CLOUD_GIT_REPO",
+            "DAGSTER_CLOUD_GIT_BRANCH",
+        ]:
+            if value := os.getenv(dagster_cloud_env_var):
+                name_clean = (
+                    dagster_cloud_env_var.lower().replace("dagster_cloud_", "").replace("_", "-")
+                )
+                tags[f"dagster/{name_clean}"] = value
+
+        return tags
+
     @public
     def get_bootstrap_env_vars(self) -> Mapping[str, str]:
         """Encode context injector and message reader params as environment variables.