feat(snowflake_hook): Adding oauth support for SnowflakeHook

.github/ISSUE_TEMPLATE/airflow_bug_report.yml

@@ -26,7 +26,7 @@ body:
       multiple: false
       options:
         - "2.10.5"
-        - "3.0.0a1"
+        - "3.0.0"
         - "main (development)"
         - "Other Airflow 2 version (please specify below)"
     validations:

INTHEWILD.md

@@ -93,6 +93,7 @@ Currently, **officially** using Airflow:
 1. [Bloomberg](https://www.techatbloomberg.com) [[@skandala23] (https://github.com/skandala23) & [@vfeldsher](https://https://github.com/vfeldsher)]
 1. [Bloomreach](https://www.bloomreach.com/) [[@neelborooah](https://github.com/neelborooah) & [@debodirno](https://github.com/debodirno) & [@ayushmnnit](https://github.com/ayushmnnit)]
 1. [Blue Yonder](http://www.blue-yonder.com) [[@blue-yonder](https://github.com/blue-yonder)]
+1. [Blue3 Investimentos](https://blue3investimentos.com.br) [[@ericcoleta] (https://github.com/ericcoleta) & [@plutaniano](https://github.com/plutaniano)]
 1. [BlueApron](https://www.blueapron.com) [[@jasonjho](https://github.com/jasonjho) & [@matthewdavidhauser](https://github.com/matthewdavidhauser)]
 1. [Bluecore](https://www.bluecore.com) [[@JLDLaughlin](https://github.com/JLDLaughlin)]
 1. [Bluekiri](https://bluekiri.com) [[@Bluekiri](https://github.com/bluekiri)]
@@ -154,7 +155,7 @@ Currently, **officially** using Airflow:
 1. [Crealytics](https://crealytics.com)
 1. [Credit Karma](https://www.creditkarma.com/) [[@preete-dixit-ck](https://github.com/preete-dixit-ck) & [@harish-gaggar-ck](https://github.com/harish-gaggar-ck) & [@greg-finley-ck](https://github.com/greg-finley-ck)]
 1. [Creditas](https://www.creditas.com.br) [[@dcassiano](https://github.com/dcassiano)]
-1. [CreditCards.com](https://www.creditcards.com/) [[@vmAggies](https://github.com/vmAggies) &  [@jay-wallaby](https://github.com/jay-wallaby)]
+1. [CreditCards.com](https://www.creditcards.com/) [[@vmAggies](https://github.com/vmAggies) & [@jay-wallaby](https://github.com/jay-wallaby)]
 1. [CRST - The Transportation Solution, Inc.](https://crst.com)
 1. [Cryptalizer.com](https://www.cryptalizer.com/)
 1. [Currency](https://www.gocurrency.com/) [[@FCLI](https://github.com/FCLI) & [@alexbegg](https://github.com/alexbegg)]
@@ -297,7 +298,7 @@ Currently, **officially** using Airflow:
 1. [Jeitto](https://www.jeitto.com.br) [[@BrennerPablo](https://github.com/BrennerPablo) & [@ds-mauri](https://github.com/ds-mauri)]
 1. [Jetlore](http://www.jetlore.com/) [[@bderose](https://github.com/bderose)]
 1. [Jobrapido](https://www.jobrapido.com/) [[@mattiagiupponi](https://github.com/mattiagiupponi)]
-1. [JobTeaser](https://www.jobteaser.com) [[@stefani75](https://github.com/stefani75) &  [@knil-sama](https://github.com/knil-sama)]
+1. [JobTeaser](https://www.jobteaser.com) [[@stefani75](https://github.com/stefani75) & [@knil-sama](https://github.com/knil-sama)]
 1. [JULO](https://www.julo.co.id/) [[@sepam](https://github.com/sepam) & [@tenapril](https://github.com/tenapril) & [@verzqy](https://github.com/verzqy)]
 1. [Kalibrr](https://www.kalibrr.com/) [[@charlesverdad](https://github.com/charlesverdad)]
 1. [Kargo](https://kargo.com) [[@chaithra-yenikapati](https://github.com/chaithra-yenikapati), [@akarsh3007](https://github.com/akarsh3007) & [@dineshanchan](https://github.com/dineshanchan)]

airflow/api_connexion/schemas/dag_schema.py

@@ -51,6 +51,8 @@ class Meta:
 
     dag_id = auto_field(dump_only=True)
     dag_display_name = fields.String(attribute="dag_display_name", dump_only=True)
+    bundle_name = auto_field(dump_only=True)
+    bundle_version = auto_field(dump_only=True)
     is_paused = auto_field()
     is_active = auto_field(dump_only=True)
     last_parsed_time = auto_field(dump_only=True)

airflow/api_fastapi/execution_api/app.py

@@ -22,12 +22,17 @@
 from typing import TYPE_CHECKING
 
 import attrs
-from fastapi import FastAPI
+from fastapi import FastAPI, Request
 from fastapi.openapi.utils import get_openapi
+from fastapi.responses import JSONResponse
 
 if TYPE_CHECKING:
     import httpx
 
+import structlog
+
+logger = structlog.get_logger(logger_name=__name__)
+
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
@@ -86,6 +91,16 @@ def custom_openapi() -> dict:
     app.openapi = custom_openapi  # type: ignore[method-assign]
 
     app.include_router(execution_api_router)
+
+    # As we are mounted as a sub app, we don't get any logs for unhandled exceptions without this!
+    @app.exception_handler(Exception)
+    def handle_exceptions(request: Request, exc: Exception):
+        logger.exception("Handle died with an error", exc_info=(type(exc), exc, exc.__traceback__))
+        content = {"message": "Internal server error"}
+        if "correlation-id" in request.headers:
+            content["correlation-id"] = request.headers["correlation-id"]
+        return JSONResponse(status_code=500, content=content)
+
     return app
 
 

airflow/cli/cli_config.py

@@ -172,6 +172,7 @@ def string_lower_type(val):
         "--bundle-name",
     ),
     help=("The name of the DAG bundle to use; may be provided more than once"),
+    type=str,
     default=None,
     action="append",
 )
@@ -638,75 +639,16 @@ def string_lower_type(val):
     default=False,
 )
 
-# webserver
-ARG_PORT = Arg(
-    ("-p", "--port"),
-    default=conf.get("webserver", "WEB_SERVER_PORT"),
-    type=int,
-    help="The port on which to run the server",
-)
-ARG_SSL_CERT = Arg(
-    ("--ssl-cert",),
-    default=conf.get("webserver", "WEB_SERVER_SSL_CERT"),
-    help="Path to the SSL certificate for the webserver",
-)
-ARG_SSL_KEY = Arg(
-    ("--ssl-key",),
-    default=conf.get("webserver", "WEB_SERVER_SSL_KEY"),
-    help="Path to the key to use with the SSL certificate",
-)
-ARG_WORKERS = Arg(
-    ("-w", "--workers"),
-    default=conf.get("webserver", "WORKERS"),
-    type=int,
-    help="Number of workers to run the webserver on",
-)
-ARG_WORKERCLASS = Arg(
-    ("-k", "--workerclass"),
-    default=conf.get("webserver", "WORKER_CLASS"),
-    choices=["sync", "eventlet", "gevent", "tornado"],
-    help="The worker class to use for Gunicorn",
-)
-ARG_WORKER_TIMEOUT = Arg(
-    ("-t", "--worker-timeout"),
-    default=conf.get("webserver", "WEB_SERVER_WORKER_TIMEOUT"),
-    type=int,
-    help="The timeout for waiting on webserver workers",
-)
-ARG_HOSTNAME = Arg(
-    ("-H", "--hostname"),
-    default=conf.get("webserver", "WEB_SERVER_HOST"),
-    help="Set the hostname on which to run the web server",
-)
-ARG_DEBUG = Arg(
-    ("-d", "--debug"), help="Use the server that ships with Flask in debug mode", action="store_true"
-)
-ARG_ACCESS_LOGFILE = Arg(
-    ("-A", "--access-logfile"),
-    default=conf.get("webserver", "ACCESS_LOGFILE"),
-    help="The logfile to store the webserver access log. Use '-' to print to stdout",
-)
-ARG_ERROR_LOGFILE = Arg(
-    ("-E", "--error-logfile"),
-    default=conf.get("webserver", "ERROR_LOGFILE"),
-    help="The logfile to store the webserver error log. Use '-' to print to stderr",
-)
-ARG_ACCESS_LOGFORMAT = Arg(
-    ("-L", "--access-logformat"),
-    default=conf.get("webserver", "ACCESS_LOGFORMAT"),
-    help="The access log format for gunicorn logs",
-)
-
 # api-server
 ARG_API_SERVER_PORT = Arg(
     ("-p", "--port"),
-    default=9091,
+    default=conf.get("api", "port"),
     type=int,
     help="The port on which to run the API server",
 )
 ARG_API_SERVER_WORKERS = Arg(
     ("-w", "--workers"),
-    default=4,
+    default=conf.get("api", "workers"),
     type=int,
     help="Number of workers to run on the API server",
 )
@@ -717,22 +659,15 @@ def string_lower_type(val):
     help="The timeout for waiting on API server workers",
 )
 ARG_API_SERVER_HOSTNAME = Arg(
-    ("-H", "--hostname"),
-    default="0.0.0.0",  # nosec
-    help="Set the hostname on which to run the API server",
+    ("-H", "--host"),
+    default=conf.get("api", "host"),
+    help="Set the host on which to run the API server",
 )
 ARG_API_SERVER_ACCESS_LOGFILE = Arg(
     ("-A", "--access-logfile"),
+    default=conf.get("api", "access_logfile"),
     help="The logfile to store the access log. Use '-' to print to stdout",
 )
-ARG_API_SERVER_ERROR_LOGFILE = Arg(
-    ("-E", "--error-logfile"),
-    help="The logfile to store the error log. Use '-' to print to stderr",
-)
-ARG_API_SERVER_ACCESS_LOGFORMAT = Arg(
-    ("-L", "--access-logformat"),
-    help="The access log format for gunicorn logs",
-)
 ARG_API_SERVER_APPS = Arg(
     ("--apps",),
     help="Applications to run (comma-separated). Default is all. Options: core, execution, all",
@@ -743,7 +678,17 @@ def string_lower_type(val):
     help="Enable X-Forwarded-Proto, X-Forwarded-For, X-Forwarded-Port to populate remote address info.",
     action="store_true",
 )
-
+ARG_SSL_CERT = Arg(
+    ("--ssl-cert",),
+    default=conf.get("api", "ssl_cert"),
+    help="Path to the SSL certificate for the webserver",
+)
+ARG_SSL_KEY = Arg(
+    ("--ssl-key",),
+    default=conf.get("api", "ssl_key"),
+    help="Path to the key to use with the SSL certificate",
+)
+ARG_DEV = Arg(("-d", "--dev"), help="Start FastAPI in development mode", action="store_true")
 
 # scheduler
 ARG_NUM_RUNS = Arg(
@@ -936,7 +881,7 @@ def string_lower_type(val):
     ("--columns",),
     type=string_list_type,
     help="List of columns to render. (default: ['dag_id', 'fileloc', 'owner', 'is_paused'])",
-    default=("dag_id", "fileloc", "owners", "is_paused"),
+    default=("dag_id", "fileloc", "owners", "is_paused", "bundle_name", "bundle_version"),
 )
 
 ARG_ASSET_LIST_COLUMNS = Arg(
@@ -1034,7 +979,7 @@ class GroupCommand(NamedTuple):
         name="list",
         help="List all the DAGs",
         func=lazy_load_command("airflow.cli.commands.remote_commands.dag_command.dag_list_dags"),
-        args=(ARG_SUBDIR, ARG_OUTPUT, ARG_VERBOSE, ARG_DAG_LIST_COLUMNS),
+        args=(ARG_OUTPUT, ARG_VERBOSE, ARG_DAG_LIST_COLUMNS, ARG_BUNDLE_NAME),
     ),
     ActionCommand(
         name="list-import-errors",
@@ -1871,13 +1816,11 @@ class GroupCommand(NamedTuple):
             ARG_STDOUT,
             ARG_STDERR,
             ARG_API_SERVER_ACCESS_LOGFILE,
-            ARG_API_SERVER_ERROR_LOGFILE,
-            ARG_API_SERVER_ACCESS_LOGFORMAT,
             ARG_API_SERVER_APPS,
             ARG_LOG_FILE,
             ARG_SSL_CERT,
             ARG_SSL_KEY,
-            ARG_DEBUG,
+            ARG_DEV,
             ARG_API_SERVER_ALLOW_PROXY_FORWARDING,
         ),
     ),

airflow/cli/commands/legacy_commands.py

@@ -48,6 +48,7 @@
     "create_user": "users create",
     "delete_user": "users delete",
     "dags backfill": "backfill create",
+    "webserver": "api-server",
 }
 
 

airflow/cli/commands/local_commands/api_server_command.py

@@ -48,13 +48,12 @@ def api_server(args):
 
     apps = args.apps
     access_logfile = args.access_logfile or "-"
-    access_logformat = args.access_logformat
     num_workers = args.workers
     worker_timeout = args.worker_timeout
     proxy_headers = args.proxy_headers
 
-    if args.debug:
-        print(f"Starting the API server on port {args.port} and host {args.hostname} debug.")
+    if args.dev:
+        print(f"Starting the API server on port {args.port} and host {args.host} in development mode.")
         log.warning("Running in dev mode, ignoring uvicorn args")
 
         run_args = [
@@ -64,13 +63,13 @@ def api_server(args):
             "--port",
             str(args.port),
             "--host",
-            str(args.hostname),
+            str(args.host),
         ]
 
         if args.proxy_headers:
             run_args.append("--proxy-headers")
 
-        # There is no way to pass the apps to airflow/api_fastapi/main.py in the debug mode
+        # There is no way to pass the apps to airflow/api_fastapi/main.py in the development mode
         # because fastapi dev command does not accept any additional arguments
         # so environment variable is being used to pass it
         os.environ["AIRFLOW_API_APPS"] = apps
@@ -91,18 +90,17 @@ def api_server(args):
                 Running the uvicorn with:
                 Apps: {apps}
                 Workers: {num_workers}
-                Host: {args.hostname}:{args.port}
+                Host: {args.host}:{args.port}
                 Timeout: {worker_timeout}
                 Logfiles: {access_logfile}
-                Access Logformat: {access_logformat}
                 ================================================================="""
             )
         )
         ssl_cert, ssl_key = _get_ssl_cert_and_key_filepaths(args)
-        setproctitle(f"airflow api_server -- host:{args.hostname} port:{args.port}")
+        setproctitle(f"airflow api_server -- host:{args.host} port:{args.port}")
         uvicorn.run(
             "airflow.api_fastapi.main:app",
-            host=args.hostname,
+            host=args.host,
             port=args.port,
             workers=num_workers,
             timeout_keep_alive=worker_timeout,

airflow/cli/commands/local_commands/standalone_command.py

@@ -203,8 +203,7 @@ def is_ready(self):
         For now, it's simply time-based.
         """
         return (
-            self.port_open(self.web_server_port)
-            and self.job_running(SchedulerJobRunner)
+            self.job_running(SchedulerJobRunner)
             and self.job_running(DagProcessorJobRunner)
             and self.job_running(TriggererJobRunner)
         )

airflow/cli/commands/remote_commands/config_command.py

@@ -292,6 +292,42 @@ def message(self) -> str:
         config=ConfigParameter("webserver", "force_log_out_after"),
         renamed_to=ConfigParameter("webserver", "session_lifetime_minutes"),
     ),
+    ConfigChange(
+        config=ConfigParameter("webserver", "web_server_host"),
+        renamed_to=ConfigParameter("api", "host"),
+    ),
+    ConfigChange(
+        config=ConfigParameter("webserver", "web_server_port"),
+        renamed_to=ConfigParameter("api", "port"),
+    ),
+    ConfigChange(
+        config=ConfigParameter("webserver", "workers"),
+        renamed_to=ConfigParameter("api", "workers"),
+    ),
+    ConfigChange(
+        config=ConfigParameter("webserver", "web_server_worker_timeout"),
+        renamed_to=ConfigParameter("api", "worker_timeout"),
+    ),
+    ConfigChange(
+        config=ConfigParameter("webserver", "web_server_ssl_cert"),
+        renamed_to=ConfigParameter("api", "ssl_cert"),
+    ),
+    ConfigChange(
+        config=ConfigParameter("webserver", "web_server_ssl_key"),
+        renamed_to=ConfigParameter("api", "ssl_key"),
+    ),
+    ConfigChange(
+        config=ConfigParameter("webserver", "access_logfile"),
+        renamed_to=ConfigParameter("api", "access_logfile"),
+    ),
+    ConfigChange(
+        config=ConfigParameter("webserver", "error_logfile"),
+        was_deprecated=False,
+    ),
+    ConfigChange(
+        config=ConfigParameter("webserver", "access_logformat"),
+        was_deprecated=False,
+    ),
     # policy
     ConfigChange(
         config=ConfigParameter("policy", "airflow_local_settings"),