3238 broader system db support

.github/workflows/test.yml

@@ -18,6 +18,12 @@ jobs:
         - { python-version: "3.7", os: "ubuntu-latest", backend-db: postgresql }
         - { python-version: "3.8", os: "ubuntu-latest", backend-db: postgresql }
         - { python-version: "3.9", os: "ubuntu-latest", backend-db: postgresql }
+        - { python-version: "3.7", os: "ubuntu-latest", backend-db: mssql }
+        - { python-version: "3.8", os: "ubuntu-latest", backend-db: mssql }
+        - { python-version: "3.9", os: "ubuntu-latest", backend-db: mssql }
+        - { python-version: "3.7", os: "ubuntu-latest", backend-db: mysql }
+        - { python-version: "3.8", os: "ubuntu-latest", backend-db: mysql }
+        - { python-version: "3.9", os: "ubuntu-latest", backend-db: mysql }
       fail-fast: false
 
     name: "Pytest on py${{ matrix.python-version }} (OS: ${{ matrix.os }}, DB: ${{ matrix.backend-db }})"
@@ -53,7 +59,7 @@ jobs:
     - name: Install Dependencies
       run: |
         poetry env use "${{ matrix.python-version }}"
-        poetry install
+        poetry install -E mysql -E mssql
 
     - name: Start Postgres Container
       if: always() && (matrix.backend-db == 'postgresql')
@@ -68,6 +74,40 @@ jobs:
         --health-retries 5
         postgres:11
 
+    - name: Start MSSQL Container
+      if: always() && (matrix.backend-db == 'mssql')
+      run: >
+        docker run -d
+        -p "1433:1433"
+        -e "SA_PASSWORD=Meltan0admin"
+        -e "ACCEPT_EULA=Y"
+        --name mssql
+        --health-cmd "/opt/mssql-tools/bin/sqlcmd -U sa -P Meltan0admin -Q 'select 1' -b -o /dev/null"
+        --health-interval 10s
+        --health-timeout 5s
+        --health-retries 5
+        mcr.microsoft.com/mssql/server:2019-latest
+
+    - name: Start MYSQL Container
+      if: always() && (matrix.backend-db == 'mysql')
+      run: >
+        docker run -d
+        -p "3306:3306"
+        -e "MYSQL_ROOT_PASSWORD=mysql"
+        --name mysql
+        --health-cmd "mysqladmin ping -h 127.0.0.1 -u root --password=mysql"
+        --health-interval 10s
+        --health-timeout 5s
+        --health-retries 5
+        mysql
+        --default-authentication-plugin=mysql_native_password
+        --collation_server=utf8_general_ci 
+        --character_set_server=utf8
+
+    - name: Wait for MYSQL Container
+      if: always() && (matrix.backend-db == 'mysql')
+      run: sleep 15
+
     - name: Check running containers
       run: |
         docker ps -a
@@ -83,6 +123,20 @@ jobs:
         POSTGRES_USER: postgres
         POSTGRES_PASSWORD: postgres
         POSTGRES_DB: pytest_warehouse
+
+        # MSSQL backend
+        MSSQL_ADDRESS: localhost
+        MSSQL_PORT: 1433
+        MSSQL_USER: sa
+        MSSQL_PASSWORD: Meltan0admin
+        MSSQL_DB: pytest_warehouse
+
+        # MySQL backend
+        MYSQL_ADDRESS: localhost
+        MYSQL_PORT: 3306
+        MYSQL_USER: root
+        MYSQL_PASSWORD: mysql
+        MYSQL_DB: pytest_warehouse
       run: |
         poetry run coverage run --parallel -m pytest -m "$PYTEST_MARKERS"
 

pyproject.toml

@@ -69,9 +69,13 @@ tzlocal = "^4.2.0"
 uvicorn = {extras = ["standard"], version = "^0.17.6"}
 watchdog = "^0.9.0"
 werkzeug = "^1"
+PyMySQL = {version = "^1.0.2", optional = true}
+pymssql = {version = "^2.2.5", optional = true}
 
 [tool.poetry.extras]
 infra = ["ansible"]
+mysql = ["pymysql"]
+mssql = ["pymssql"]
 
 [tool.poetry.dev-dependencies]
 asynctest = "^0.12.2"

src/meltano/migrations/utils/dialect_typing.py

@@ -0,0 +1,35 @@
+from typing import Optional, Union
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects.mssql import DATETIME2
+from sqlalchemy.engine import Connection
+
+
+def get_dialect_name(op: op) -> str:
+    """
+    Get the dialect name from the Alembic op.
+    """
+    database_connection: Connection = op.get_bind()
+    return database_connection.dialect.name
+
+def datetime_for_dialect(dialect_name: str) -> Union[DATETIME2, sa.DateTime]:
+    """
+    Get the datetime type for the given dialect.
+    """
+    # We need to use the DATETIME2 type for MSSQL, because the 
+    # default DATETIME type does not go back to the year 1.
+    if dialect_name == "mssql":
+        return DATETIME2
+
+    return sa.DateTime
+
+def max_string_length_for_dialect(dialect_name: str) -> Optional[int]:
+    """
+    Get the maximum string length for the given dialect.
+    We need to limit the size of the string to avoid MySQL throwing an error.
+    """
+    if dialect_name == "mysql":
+        return 1024
+
+    return None

src/meltano/migrations/versions/13e8639c6d2b_add_state_edit_to_job_state_enum.py

@@ -10,6 +10,11 @@
 import sqlalchemy as sa
 from alembic import op
 
+from meltano.migrations.utils.dialect_typing import (
+    get_dialect_name,
+    max_string_length_for_dialect,
+)
+
 # revision identifiers, used by Alembic.
 revision = "13e8639c6d2b"
 down_revision = "d135f52a6f49"
@@ -30,14 +35,17 @@ class State(Enum):
 
 
 def upgrade():
+    dialect_name = get_dialect_name(op)
+    max_string_length = max_string_length_for_dialect(dialect_name)
+
     conn = op.get_bind()
     # In sqlite, the field is already a varchar.
     # "ALTER COLUMN" statements are also not supported.
     if conn.dialect.name != "sqlite":
         op.alter_column(
             table_name="job",
             column_name="state",
-            type_=sa.types.String,
+            type_=sa.types.String(max_string_length),
             existing_type=sa.Enum(State, name="job_state"),
         )
 

src/meltano/migrations/versions/23ea52e6d784_add_resource_type_to_embed_token.py

@@ -8,6 +8,11 @@
 import sqlalchemy as sa
 from alembic import op
 
+from meltano.migrations.utils.dialect_typing import (
+    get_dialect_name,
+    max_string_length_for_dialect,
+)
+
 # revision identifiers, used by Alembic.
 revision = "23ea52e6d784"
 down_revision = "ceb00d7ff3bd"
@@ -18,7 +23,10 @@
 
 
 def upgrade():
-    op.add_column("embed_tokens", sa.Column("resource_type", sa.String()))
+    dialect_name = get_dialect_name(op)
+    max_string_length = max_string_length_for_dialect(dialect_name)
+
+    op.add_column("embed_tokens", sa.Column("resource_type", sa.String(max_string_length)))
 
     metadata = sa.MetaData(bind=op.get_bind())
     Embed_Tokens = sa.Table("embed_tokens", metadata, autoload=True)

src/meltano/migrations/versions/367228df6a43_add_trigger_to_job.py

@@ -8,6 +8,11 @@
 import sqlalchemy as sa
 from alembic import op
 
+from meltano.migrations.utils.dialect_typing import (
+    get_dialect_name,
+    max_string_length_for_dialect,
+)
+
 # revision identifiers, used by Alembic.
 revision = "367228df6a43"
 down_revision = "87d9638f6ac6"
@@ -16,7 +21,10 @@
 
 
 def upgrade():
-    op.add_column("job", sa.Column("trigger", sa.String()))
+    dialect_name = get_dialect_name(op)
+    max_string_length = max_string_length_for_dialect(dialect_name)
+
+    op.add_column("job", sa.Column("trigger", sa.String(max_string_length)))
 
 
 def downgrade():

src/meltano/migrations/versions/6ef30ab7b8e5_.py

@@ -8,6 +8,12 @@
 import sqlalchemy as sa
 from alembic import op
 
+from meltano.migrations.utils.dialect_typing import (
+    datetime_for_dialect,
+    get_dialect_name,
+    max_string_length_for_dialect,
+)
+
 # revision identifiers, used by Alembic.
 revision = "6ef30ab7b8e5"
 down_revision = "b4c05e463b53"
@@ -16,14 +22,18 @@
 
 
 def upgrade():
+    dialect_name = get_dialect_name(op)
+    datetime_type = datetime_for_dialect(dialect_name)
+    max_string_length = max_string_length_for_dialect(dialect_name)
+
     op.create_table(
         "user",
         sa.Column("id", sa.Integer, nullable=False),
-        sa.Column("username", sa.String),
-        sa.Column("email", sa.String),
-        sa.Column("password", sa.String),
+        sa.Column("username", sa.String(255)),
+        sa.Column("email", sa.String(255)),
+        sa.Column("password", sa.String(255)),
         sa.Column("active", sa.Boolean),
-        sa.Column("confirmed_at", sa.DateTime, nullable=True),
+        sa.Column("confirmed_at", datetime_type, nullable=True),
         sa.PrimaryKeyConstraint("id"),
         sa.UniqueConstraint("email"),
     )
@@ -33,7 +43,7 @@ def upgrade():
         "role",
         sa.Column("id", sa.Integer, nullable=False),
         sa.Column("name", sa.String(80)),
-        sa.Column("description", sa.String, nullable=True),
+        sa.Column("description", sa.String(255), nullable=True),
         sa.PrimaryKeyConstraint("id"),
         sa.UniqueConstraint("name"),
     )
@@ -52,8 +62,8 @@ def upgrade():
         "role_permissions",
         sa.Column("id", sa.Integer, nullable=False),
         sa.Column("role_id", sa.Integer),
-        sa.Column("type", sa.String),
-        sa.Column("context", sa.String, nullable=True),
+        sa.Column("type", sa.String(max_string_length)),
+        sa.Column("context", sa.String(max_string_length), nullable=True),
         sa.ForeignKeyConstraint(["role_id"], ["role.id"]),
         sa.PrimaryKeyConstraint("id"),
     )
@@ -65,8 +75,8 @@ def upgrade():
         sa.Column("provider_id", sa.String(255)),
         sa.Column("provider_user_id", sa.Integer, nullable=True),
         sa.Column("access_token", sa.String(255)),
-        sa.Column("created_at", sa.DateTime, nullable=True),
-        sa.Column("id_token", sa.String),
+        sa.Column("created_at", datetime_type, nullable=True),
+        sa.Column("id_token", sa.String(max_string_length)),
         sa.ForeignKeyConstraint(["user_id"], ["user.id"]),
         sa.PrimaryKeyConstraint("id"),
     )

src/meltano/migrations/versions/87d9638f6ac6_add_subscription.py

@@ -7,7 +7,12 @@
 """
 import sqlalchemy as sa
 from alembic import op
+
 from meltano.migrations import GUID
+from meltano.migrations.utils.dialect_typing import (
+    datetime_for_dialect,
+    get_dialect_name,
+)
 
 # revision identifiers, used by Alembic.
 revision = "87d9638f6ac6"
@@ -17,14 +22,17 @@
 
 
 def upgrade():
+    dialect_name = get_dialect_name(op)
+    datetime_type = datetime_for_dialect(dialect_name)
+
     op.create_table(
         "subscriptions",
         sa.Column("id", GUID, primary_key=True),
-        sa.Column("recipient", sa.String(), nullable=False),
-        sa.Column("event_type", sa.String(), nullable=False),
-        sa.Column("source_type", sa.String(), nullable=True),
-        sa.Column("source_id", sa.String(), nullable=True),
-        sa.Column("created_at", sa.DateTime),
+        sa.Column("recipient", sa.String(255), nullable=False),
+        sa.Column("event_type", sa.String(255), nullable=False),
+        sa.Column("source_type", sa.String(255), nullable=True),
+        sa.Column("source_id", sa.String(255), nullable=True),
+        sa.Column("created_at", datetime_type),
         sa.UniqueConstraint("recipient", "event_type", "source_type", "source_id"),
     )
 

src/meltano/migrations/versions/a3e2b0a4937d_add_login_audit_columns.py

@@ -8,6 +8,11 @@
 import sqlalchemy as sa
 from alembic import op
 
+from meltano.migrations.utils.dialect_typing import (
+    datetime_for_dialect,
+    get_dialect_name,
+)
+
 # revision identifiers, used by Alembic.
 revision = "a3e2b0a4937d"
 down_revision = "53e97221d99f"
@@ -16,7 +21,10 @@
 
 
 def upgrade():
-    op.add_column("user", sa.Column("last_login_at", sa.DateTime(), nullable=True))
+    dialect_name = get_dialect_name(op)
+    datetime_type = datetime_for_dialect(dialect_name)
+
+    op.add_column("user", sa.Column("last_login_at", datetime_type, nullable=True))
     op.add_column("user", sa.Column("login_count", sa.Integer, default=0))
 
 

src/meltano/migrations/versions/b4c05e463b53_.py

@@ -9,9 +9,16 @@
 
 import sqlalchemy as sa
 from alembic import op
-from meltano.migrations import IntFlag, JSONEncodedDict
+from sqlalchemy.dialects.mssql import DATETIME2
 from sqlalchemy.ext.mutable import MutableDict
 
+from meltano.migrations import IntFlag, JSONEncodedDict
+from meltano.migrations.utils.dialect_typing import (
+    datetime_for_dialect,
+    get_dialect_name,
+    max_string_length_for_dialect,
+)
+
 # revision identifiers, used by Alembic.
 revision = "b4c05e463b53"
 down_revision = None
@@ -29,23 +36,27 @@ class State(Enum):
 
 
 def upgrade():
+    dialect_name = get_dialect_name(op)
+    datetime_type = datetime_for_dialect(dialect_name)
+    max_string_length = max_string_length_for_dialect(dialect_name)
+
     op.create_table(
         "job",
         sa.Column("id", sa.Integer, primary_key=True),
-        sa.Column("job_id", sa.String),
+        sa.Column("job_id", sa.String(max_string_length)),
         sa.Column("state", sa.Enum(State, name="job_state")),
-        sa.Column("started_at", sa.DateTime),
-        sa.Column("ended_at", sa.DateTime),
-        sa.Column("payload", MutableDict.as_mutable(JSONEncodedDict)),
+        sa.Column("started_at", datetime_type),
+        sa.Column("ended_at", datetime_type),
+        sa.Column("payload", MutableDict.as_mutable(JSONEncodedDict(max_string_length))),
         sa.Column("payload_flags", IntFlag, default=0),
     )
 
     op.create_table(
         "plugin_settings",
-        sa.Column("label", sa.String(), nullable=True),
+        sa.Column("label", datetime_type, nullable=True),
         sa.Column("description", sa.Text(), nullable=True),
-        sa.Column("name", sa.String(), nullable=False),
-        sa.Column("namespace", sa.String(), nullable=True),
+        sa.Column("name", sa.String(255), nullable=False),
+        sa.Column("namespace", sa.String(255), nullable=True),
         sa.Column("value", sa.PickleType(), nullable=True),
         sa.Column("enabled", sa.Boolean(), nullable=True),
         sa.PrimaryKeyConstraint("name", "namespace"),