feat: use sqlalchemy 1.4 instead of 2 (#8)

* feat: use sqlalchemy 1.4 instead of 2

* define default targets

Author: aquamatthias

cloud2sql/__main__.py

@@ -2,7 +2,8 @@
 
 from resotolib.args import Namespace, ArgumentParser
 from resotolib.logger import setup_logger
-from sqlalchemy import create_engine, Engine
+from sqlalchemy import create_engine
+from sqlalchemy.engine import Engine
 
 from cloud2sql.collect_plugins import collect_from_plugins
 

cloud2sql/collect_plugins.py

@@ -22,7 +22,7 @@
 from resotolib.types import Json
 from rich import print as rich_print
 from rich.live import Live
-from sqlalchemy import Engine
+from sqlalchemy.engine import Engine
 
 from cloud2sql.show_progress import CollectInfo
 from cloud2sql.sql import SqlModel, SqlUpdater
@@ -64,40 +64,41 @@ def collect(collector: BaseCollectorPlugin, engine: Engine, feedback: CoreFeedba
     kinds = [from_json(m, Kind) for m in collector.graph.export_model(walk_subclasses=False)]
     model = SqlModel(Model({k.fqn: k for k in kinds}))
     node_edge_count = len(collector.graph.nodes) + len(collector.graph.edges)
-    ne_current = 0
-    progress_update = 5000
-    feedback.progress_done("sync_db", 0, node_edge_count, context=[collector.cloud])
+    ne_count = iter(range(0, node_edge_count))
+    progress_update = max(node_edge_count // 100, 50)
+    schema = f"create temp tables {engine.dialect.name}"
+    syncdb = f"synchronize {engine.dialect.name}"
+    feedback.progress_done(schema, 0, 1, context=[collector.cloud])
+    feedback.progress_done(syncdb, 0, node_edge_count, context=[collector.cloud])
     with engine.connect() as conn:
-        # create the ddl metadata from the kinds
-        model.create_schema(conn, args)
-        # ingest the data
-        updater = SqlUpdater(model)
-        node: BaseResource
-        for node in collector.graph.nodes:
-            node._graph = collector.graph
-            exported = node_to_dict(node)
-            exported["type"] = "node"
-            exported["ancestors"] = {
-                "cloud": {"reported": {"id": node.cloud().name}},
-                "account": {"reported": {"id": node.account().name}},
-                "region": {"reported": {"id": node.region().name}},
-                "zone": {"reported": {"id": node.zone().name}},
-            }
-            stmt = updater.insert_node(exported)
-            if stmt is not None:
-                conn.execute(stmt)
-            ne_current += 1
-            if ne_current % progress_update == 0:
-                feedback.progress_done("sync_db", ne_current, node_edge_count, context=[collector.cloud])
-        for from_node, to_node, _ in collector.graph.edges:
-            stmt = updater.insert_node({"from": from_node.chksum, "to": to_node.chksum, "type": "edge"})
-            if stmt is not None:
-                conn.execute(stmt)
-            ne_current += 1
-            if ne_current % progress_update == 0:
-                feedback.progress_done("sync_db", ne_current, node_edge_count, context=[collector.cloud])
-        # commit all the changes to the tmp tables
-        conn.commit()
+        with conn.begin():
+            # create the ddl metadata from the kinds
+            model.create_schema(conn, args)
+            feedback.progress_done(schema, 1, 1, context=[collector.cloud])
+            # ingest the data
+            updater = SqlUpdater(model)
+            node: BaseResource
+            for node in collector.graph.nodes:
+                node._graph = collector.graph
+                exported = node_to_dict(node)
+                exported["type"] = "node"
+                exported["ancestors"] = {
+                    "cloud": {"reported": {"id": node.cloud().name}},
+                    "account": {"reported": {"id": node.account().name}},
+                    "region": {"reported": {"id": node.region().name}},
+                    "zone": {"reported": {"id": node.zone().name}},
+                }
+                stmt = updater.insert_node(exported)
+                if stmt is not None:
+                    conn.execute(stmt)
+                if (nx := next(ne_count)) % progress_update == 0:
+                    feedback.progress_done(syncdb, nx, node_edge_count, context=[collector.cloud])
+            for from_node, to_node, _ in collector.graph.edges:
+                stmt = updater.insert_node({"from": from_node.chksum, "to": to_node.chksum, "type": "edge"})
+                if stmt is not None:
+                    conn.execute(stmt)
+                if (nx := next(ne_count)) % progress_update == 0:
+                    feedback.progress_done(syncdb, nx, node_edge_count, context=[collector.cloud])
     feedback.progress_done(collector.cloud, 1, 1)
 
 

cloud2sql/collect_resoto.py

@@ -1,6 +1,6 @@
 from resotoclient import ResotoClient
 from resotolib.args import Namespace
-from sqlalchemy import Engine
+from sqlalchemy.engine import Engine
 
 from cloud2sql.sql import SqlModel, SqlUpdater
 
@@ -12,9 +12,9 @@ def collect_from_resoto(engine: Engine, args: Namespace) -> None:
         updater = SqlUpdater(model)
 
         with engine.connect() as conn:
-            model.create_schema(conn, args)
-            for nd in client.search_graph("id(root) -[0:]->"):
-                stmt = updater.insert_node(nd)
-                if stmt is not None:
-                    conn.execute(stmt)
-            conn.commit()
+            with conn.begin():
+                model.create_schema(conn, args)
+                for nd in client.search_graph("id(root) -[0:]->"):
+                    stmt = updater.insert_node(nd)
+                    if stmt is not None:
+                        conn.execute(stmt)

cloud2sql/sql.py

@@ -7,23 +7,11 @@
 from resotolib.args import Namespace
 from resotolib.baseresources import BaseResource
 from resotolib.types import Json
-from sqlalchemy import (
-    Boolean,
-    Column,
-    Connection,
-    Double,
-    Float,
-    Insert,
-    Integer,
-    JSON,
-    MetaData,
-    String,
-    Table,
-    ValuesBase,
-    DDL,
-    Engine,
-)
+from sqlalchemy import Boolean, Column, Float, Integer, JSON, MetaData, String, Table, DDL
+from sqlalchemy.engine import Engine, Connection
+from sqlalchemy.sql import Insert
 from sqlalchemy.sql.ddl import DropTable, DropConstraint
+from sqlalchemy.sql.dml import ValuesBase
 from sqlalchemy.sql.type_api import TypeEngine
 
 from cloud2sql.util import value_in_path
@@ -66,7 +54,7 @@ def column_type_from(kind: str) -> Type[TypeEngine[Any]]:
         elif kind in "float":
             return Float
         elif kind in "double":
-            return Double
+            return Float  # use Double with sqlalchemy 2
         elif kind in ("string", "date", "datetime", "duration"):
             return String
         elif kind == "boolean":
@@ -159,22 +147,22 @@ def link_table_schema_from_successors(kind: Kind) -> None:
     @staticmethod
     def swap_temp_tables(engine: Engine) -> None:
         with engine.connect() as connection:
-            metadata = MetaData()
-            metadata.reflect(connection, resolve_fks=False)
-
-            def drop_table(tl: Table) -> None:
-                for cs in tl.foreign_key_constraints:
-                    connection.execute(DropConstraint(cs))  # type: ignore
-                connection.execute(DropTable(tl))
-
-            for table in metadata.tables.values():
-                if table.name.startswith(temp_prefix):
-                    prod_table = table.name[len(temp_prefix) :]  # noqa: E203
-                    if prod_table in metadata.tables:
-                        drop_table(metadata.tables[prod_table])
-                    connection.execute(DDL(f"ALTER TABLE {table.name} RENAME TO {prod_table}"))  # type: ignore
-            # todo: create foreign key constraints on the final tables
-            connection.commit()
+            with connection.begin():
+                metadata = MetaData()
+                metadata.reflect(connection, resolve_fks=False)
+
+                def drop_table(tl: Table) -> None:
+                    for cs in tl.foreign_key_constraints:
+                        connection.execute(DropConstraint(cs))  # type: ignore
+                    connection.execute(DropTable(tl))
+
+                for table in metadata.tables.values():
+                    if table.name.startswith(temp_prefix):
+                        prod_table = table.name[len(temp_prefix) :]  # noqa: E203
+                        if prod_table in metadata.tables:
+                            drop_table(metadata.tables[prod_table])
+                        connection.execute(DDL(f"ALTER TABLE {table.name} RENAME TO {prod_table}"))  # type: ignore
+                # todo: create foreign key constraints on the final tables
 
 
 class SqlUpdater:

requirements.txt

@@ -1,7 +1,7 @@
 # Only list production dependencies. Everything else can be added to requirements-test.txt and requirements-dev.txt.
 
 # required for cloud2sql to run
-SQLAlchemy==2.0.0b4
+SQLAlchemy==1.4.45
 PyYAML>=6.0
 rich>=12.6.0
 resotoclient>=1.2.1

setup.cfg

@@ -1,8 +1,7 @@
 [flake8]
 max-line-length = 120
 exclude = .git,.tox,__pycache__,.idea,.pytest_cache,docs
-application-import-names = resotocore tests
-ignore = E1131, N818, W503
+application-import-names = cloud2sql tests
 
 [tool:pytest]
 log_cli=true

tests/conftest.py

@@ -6,7 +6,7 @@
 from resotolib.args import Namespace
 from resotolib.core.actions import CoreFeedback
 from resotolib.types import Json
-from sqlalchemy import create_engine, Engine
+from sqlalchemy.engine import create_engine, Engine
 
 from cloud2sql.sql import SqlModel, SqlUpdater
 

tests/sql_test.py

@@ -1,6 +1,7 @@
 from resotoclient.models import Model
 from resotolib.args import Namespace
-from sqlalchemy import Engine, MetaData
+from sqlalchemy import MetaData
+from sqlalchemy.engine import Engine
 from sqlalchemy.orm import Session
 
 from cloud2sql.sql import SqlModel, SqlUpdater

tox.ini

@@ -1,3 +1,10 @@
+[tox]
+env_list = 
+  tests
+  syntax
+  black
+  mypy
+
 [pytest]
 testpaths= tests