feat: split database information (#24067)

superset/databases/api.py

@@ -61,6 +61,7 @@ from superset.databases.filters import DatabaseFilter, DatabaseUploadEnabledFilt
 from superset.databases.schemas import (
     database_schemas_query_schema,
     database_tables_query_schema,
+    DatabaseConnectionSchema,
     DatabaseFunctionNamesResponse,
     DatabasePostSchema,
     DatabasePutSchema,
@@ -122,6 +123,7 @@ class DatabaseRestApi(BaseSupersetModelRestApi):
         "validate_sql",
         "delete_ssh_tunnel",
         "schemas_access_for_file_upload",
+        "get_connection",
     }
     resource_name = "database"
     class_permission_name = "Database"
@@ -144,12 +146,6 @@ class DatabaseRestApi(BaseSupersetModelRestApi):
         "driver",
         "force_ctas_schema",
         "impersonate_user",
-        "masked_encrypted_extra",
-        "extra",
-        "parameters",
-        "parameters_schema",
-        "server_cert",
-        "sqlalchemy_uri",
         "is_managed_externally",
         "engine_information",
     ]
@@ -223,6 +219,7 @@ class DatabaseRestApi(BaseSupersetModelRestApi):
 
     openapi_spec_tag = "Database"
     openapi_spec_component_schemas = (
+        DatabaseConnectionSchema,
         DatabaseFunctionNamesResponse,
         DatabaseSchemaAccessForFileUploadResponse,
         DatabaseRelatedObjectsResponse,
@@ -237,6 +234,50 @@ class DatabaseRestApi(BaseSupersetModelRestApi):
         ValidateSQLResponse,
     )
 
+    @expose("/<int:pk>/connection", methods=("GET",))
+    @protect()
+    @safe
+    def get_connection(self, pk: int) -> Response:
+        """Get database connection info.
+        ---
+        get:
+          summary: >-
+            Get a database connection info
+          parameters:
+          - in: path
+            schema:
+              type: integer
+            description: The database id
+            name: pk
+          responses:
+            200:
+              description: Database with connection info
+              content:
+                application/json:
+                  schema:
+                    $ref: "#/components/schemas/DatabaseConnectionSchema"
+            400:
+              $ref: '#/components/responses/400'
+            401:
+              $ref: '#/components/responses/401'
+            422:
+              $ref: '#/components/responses/422'
+            500:
+              $ref: '#/components/responses/500'
+        """
+        database = DatabaseDAO.find_by_id(pk)
+        database_connection_schema = DatabaseConnectionSchema()
+        response = {
+            "id": pk,
+            "result": database_connection_schema.dump(database, many=False),
+        }
+        try:
+            if ssh_tunnel := DatabaseDAO.get_ssh_tunnel(pk):
+                response["result"]["ssh_tunnel"] = ssh_tunnel.data
+            return self.response(200, **response)
+        except SupersetException as ex:
+            return self.response(ex.status, message=ex.message)
+
     @expose("/<int:pk>", methods=("GET",))
     @protect()
     @safe

superset/databases/schemas.py

@@ -880,3 +880,86 @@ class DatabaseSchemaAccessForFileUploadResponse(Schema):
             "information"
         },
     )
+
+
+class DatabaseConnectionSchema(Schema):
+    """
+    Schema with database connection information.
+
+    This is only for admins (who have ``can_create`` on ``Database``).
+    """
+
+    allow_ctas = fields.Boolean(metadata={"description": allow_ctas_description})
+    allow_cvas = fields.Boolean(metadata={"description": allow_cvas_description})
+    allow_dml = fields.Boolean(metadata={"description": allow_dml_description})
+    allow_file_upload = fields.Boolean(
+        metadata={"description": allow_file_upload_description}
+    )
+    allow_run_async = fields.Boolean(
+        metadata={"description": allow_run_async_description}
+    )
+    backend = fields.String(
+        allow_none=True, metadata={"description": "SQLAlchemy engine to use"}
+    )
+    cache_timeout = fields.Integer(
+        metadata={"description": cache_timeout_description}, allow_none=True
+    )
+    configuration_method = fields.String(
+        metadata={"description": configuration_method_description},
+    )
+    database_name = fields.String(
+        metadata={"description": database_name_description},
+        allow_none=True,
+        validate=Length(1, 250),
+    )
+    driver = fields.String(
+        allow_none=True, metadata={"description": "SQLAlchemy driver to use"}
+    )
+    engine_information = fields.Dict(keys=fields.String(), values=fields.Raw())
+    expose_in_sqllab = fields.Boolean(
+        metadata={"description": expose_in_sqllab_description}
+    )
+    extra = fields.String(
+        metadata={"description": extra_description}, validate=extra_validator
+    )
+    force_ctas_schema = fields.String(
+        metadata={"description": force_ctas_schema_description},
+        allow_none=True,
+        validate=Length(0, 250),
+    )
+    id = fields.Integer(metadata={"description": "Database ID (for updates)"})
+    impersonate_user = fields.Boolean(
+        metadata={"description": impersonate_user_description}
+    )
+    is_managed_externally = fields.Boolean(allow_none=True, dump_default=False)
+    server_cert = fields.String(
+        metadata={"description": server_cert_description},
+        allow_none=True,
+        validate=server_cert_validator,
+    )
+    uuid = fields.String(required=False)
+    ssh_tunnel = fields.Nested(DatabaseSSHTunnel, allow_none=True)
+    masked_encrypted_extra = fields.String(
+        metadata={"description": encrypted_extra_description},
+        validate=encrypted_extra_validator,
+        allow_none=True,
+    )
+    parameters = fields.Dict(
+        keys=fields.String(),
+        values=fields.Raw(),
+        metadata={"description": "DB-specific parameters for configuration"},
+    )
+    parameters_schema = fields.Dict(
+        keys=fields.String(),
+        values=fields.Raw(),
+        metadata={
+            "description": (
+                "JSONSchema for configuring the database by "
+                "parameters instead of SQLAlchemy URI"
+            ),
+        },
+    )
+    sqlalchemy_uri = fields.String(
+        metadata={"description": sqlalchemy_uri_description},
+        validate=[Length(1, 1024), sqlalchemy_uri_validator],
+    )