turbaszek commented on a change in pull request #12704:
URL: https://github.com/apache/airflow/pull/12704#discussion_r532262483
##########
File path: airflow/cli/simple_table.py
##########
@@ -14,11 +14,76 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+import json
+from typing import Any, Callable, Dict, List, Optional, Union
+import yaml
from rich.box import ASCII_DOUBLE_HEAD
+from rich.console import Console
+from rich.syntax import Syntax
from rich.table import Table
+class AirflowConsole(Console):
+ """Airflow rich console"""
+
+ def print_as_json(self, data: Dict):
+ """Renders dict as json text representation"""
+ json_content = json.dumps(data)
+ self.print(Syntax(json_content, "json", theme="ansi_dark"),
soft_wrap=True)
+
+ def print_as_yaml(self, data: Dict):
+ """Renders dict as yaml text representation"""
+ yaml_content = yaml.dump(data)
+ self.print(Syntax(yaml_content, "yaml", theme="ansi_dark"),
soft_wrap=True)
+
+ def print_as_table(self, data: List[Dict]):
+ """Renders list of dictionaries as table"""
+ if not data:
+ self.print("No data found")
+ return
+
+ table = SimpleTable(
+ show_header=True,
+ )
+ for col in data[0].keys():
+ table.add_column(col)
+
+ for row in data:
+ table.add_row(*[str(d) for d in row.values()])
+ self.print(table)
+
+ def _normalize_data(self, value: Any, output: str) -> Union[list, str,
dict]:
+ if isinstance(value, (tuple, list)):
+ if output == "table":
+ return ",".join(self._normalize_data(x, output) for x in value)
+ return [self._normalize_data(x, output) for x in value]
+ if isinstance(value, dict) and output != "table":
+ return {k: self._normalize_data(v, output) for k, v in
value.items()}
+ return str(value)
+
+ def print_as(self, data: List[Union[Dict, Any]], output: str, mapper:
Optional[Callable] = None):
+ """Prints provided using format specified by output argument"""
+ output_to_rendered = {
+ "json": self.print_as_json,
+ "yaml": self.print_as_yaml,
+ "table": self.print_as_table,
+ }
+ renderer = output_to_rendered.get(output)
+ if not renderer:
+ raise ValueError("The output")
Review comment:
Oh, yeah, definitely xD
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
