[database] Improve form and API validation for invalid URI (#8240)

* [database] Improve form and API validation for invalid URI

* [database] Added missing EOL

* [database] lint

superset/views/database/__init__.py

@@ -16,10 +16,14 @@
 # under the License.
 # pylint: disable=C,R,W
 import inspect
+from typing import Type
 
 from flask import Markup
 from flask_babel import lazy_gettext as _
+from marshmallow import ValidationError
 from sqlalchemy import MetaData
+from sqlalchemy.engine.url import make_url
+from sqlalchemy.exc import ArgumentError
 
 from superset import security_manager
 from superset.exceptions import SupersetException
@@ -27,6 +31,24 @@ from superset.utils import core as utils
 from superset.views.base import SupersetFilter
 
 
+def sqlalchemy_uri_validator(
+    uri: str, exception: Type[ValidationError] = ValidationError
+) -> None:
+    """
+    Check if a user has submitted a valid SQLAlchemy URI
+    """
+    try:
+        make_url(uri.strip())
+    except ArgumentError:
+        raise exception(
+            _(
+                "Invalid connnection string, a valid string follows: "
+                " 'DRIVER://USER:PASSWORD@DB-HOST/DATABASE-NAME'"
+                " <p>Example:'postgresql://user:password@your-postgres-db/database'</p>"
+            )
+        )
+
+
 class DatabaseFilter(SupersetFilter):
     def apply(self, query, func):  # noqa
         if security_manager.all_database_access():
@@ -189,7 +211,7 @@ class DatabaseMixin:  # noqa
         "backend": _("Backend"),
     }
 
-    def pre_add(self, db):
+    def _pre_add_update(self, db):
         self.check_extra(db)
         db.set_sqlalchemy_uri(db.sqlalchemy_uri)
         security_manager.add_permission_view_menu("database_access", db.perm)
@@ -199,8 +221,11 @@ class DatabaseMixin:  # noqa
                 "schema_access", security_manager.get_schema_perm(db, schema)
             )
 
+    def pre_add(self, db):
+        self._pre_add_update(db)
+
     def pre_update(self, db):
-        self.pre_add(db)
+        self._pre_add_update(db)
 
     def pre_delete(self, obj):
         if obj.tables: