Remove sqlalchemy-utils

CHANGELOG.md

@@ -1,5 +1,14 @@
 # Changelog
 
+## [1.1.0]
+
+10/12/2025
+
+- Remove dependency on sqlalchemy-utils.
+- Add our own database utility functions for test purposes.
+- Add comprehensive test suite for database utility functions.
+
+
 ## [1.0.4]
 
 18/11/2025

README.md

@@ -6,7 +6,6 @@ A tool for comparing database schemas using SQLAlchemy.
 
 - Python 3.10 or higher (supports 3.10, 3.11, 3.12, 3.13, 3.14)
 - SQLAlchemy >= 1.4
-- sqlalchemy-utils ~= 0.41.2
 
 ## Authors
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "sqlalchemy-diff"
-version = "1.0.4"
+version = "1.1.0"
 authors = [
     { name = "Fabrizio Romano", email = "[email protected]" },
     { name = "Mark McArdle", email = "[email protected]" },
@@ -31,7 +31,6 @@ maintainers = [
 keywords = ["sqlalchemy", "diff", "compare"]
 dependencies = [
     "sqlalchemy>=1.4,<3",
-    "sqlalchemy-utils>=0.40.0,!=0.42.0",
 ]
 
 [project.optional-dependencies]

src/sqlalchemydiff/inspection/base.py

@@ -1,5 +1,6 @@
 import abc
 import inspect as stdlib_inspect
+from typing import Any
 
 from sqlalchemy import inspect
 from sqlalchemy.engine import Engine
@@ -59,7 +60,7 @@ def __init__(self, one_alias: str = "one", two_alias: str = "two"):
     @abc.abstractmethod
     def inspect(
         self, engine: Engine, ignore_specs: list[IgnoreSpecType] | None = None
-    ) -> dict: ...  # pragma: no cover
+    ) -> Any: ...  # pragma: no cover
 
     @abc.abstractmethod
     def diff(self, one: dict, two: dict) -> dict: ...  # pragma: no cover

src/sqlalchemydiff/inspection/inspectors.py

@@ -1,4 +1,4 @@
-from typing import cast
+from typing import Any, cast
 
 from sqlalchemy.engine import Engine
 
@@ -144,7 +144,7 @@ def diff(self, one: dict, two: dict) -> dict:
     def _is_supported(self, inspector: Inspector) -> bool:
         return hasattr(inspector, "get_foreign_keys")
 
-    def _get_fk_identifier(self, fk: dict) -> dict:
+    def _get_fk_identifier(self, fk: Any) -> Any:
         if not fk["name"]:
             fk["name"] = f"_unnamed_fk_{fk['referred_table']}_{'_'.join(fk['constrained_columns'])}"
         return fk

tests/db_util.py

@@ -0,0 +1,173 @@
+from pathlib import Path
+from urllib.parse import urlparse
+
+from sqlalchemy import create_engine, text
+
+
+def _parse_database_uri(uri):
+    """Parse a database URI and return components."""
+    parsed = urlparse(uri)
+    return {
+        "scheme": parsed.scheme,
+        "username": parsed.username,
+        "password": parsed.password,
+        "hostname": parsed.hostname,
+        "port": parsed.port,
+        "database": parsed.path.lstrip("/") if parsed.path else None,
+        "path": parsed.path,
+    }
+
+
+def _get_postgres_admin_uri(uri):
+    """Get a PostgreSQL URI pointing to the 'postgres' database for admin operations."""
+    parsed = urlparse(uri)
+    return parsed._replace(path="postgres").geturl()
+
+
+def _get_sqlite_file_path(uri):
+    """Extract the file path from a SQLite URI as a Path object."""
+    parsed = _parse_database_uri(uri)
+    path = parsed["path"]
+
+    # Handle :memory: and empty path
+    if not path or path == "/:memory:" or path == "/":
+        return None
+
+    # Remove leading slash
+    if path.startswith("/"):
+        path = path[1:]
+
+    # Handle empty string after stripping
+    if not path:
+        return None
+
+    return Path(path)
+
+
+def database_exists(uri):
+    """Check if a database exists.
+
+    Args:
+        uri: Database URI (postgresql://... or sqlite:///...)
+
+    Returns:
+        bool: True if database exists, False otherwise
+    """
+    parsed = _parse_database_uri(uri)
+    scheme = parsed["scheme"]
+
+    if scheme == "postgresql":
+        admin_uri = _get_postgres_admin_uri(uri)
+        target_db = parsed["database"]
+
+        engine = create_engine(admin_uri, isolation_level="AUTOCOMMIT")
+        try:
+            with engine.connect() as conn:
+                result = conn.execute(
+                    text("SELECT 1 FROM pg_database WHERE datname = :dbname"), {"dbname": target_db}
+                )
+                return result.fetchone() is not None
+        finally:
+            engine.dispose()
+
+    elif scheme == "sqlite":
+        file_path = _get_sqlite_file_path(uri)
+
+        # In-memory databases always "exist" (they're created on connection)
+        if file_path is None:
+            return True
+
+        # For file-based SQLite, check if file exists
+        return file_path.exists()
+
+    else:
+        raise ValueError(f"Unsupported database scheme: {scheme}")
+
+
+def create_database(uri):
+    """Create a database.
+
+    Args:
+        uri: Database URI (postgresql://... or sqlite:///...)
+    """
+    parsed = _parse_database_uri(uri)
+    scheme = parsed["scheme"]
+
+    if scheme == "postgresql":
+        admin_uri = _get_postgres_admin_uri(uri)
+        target_db = parsed["database"]
+
+        engine = create_engine(admin_uri, isolation_level="AUTOCOMMIT")
+        try:
+            with engine.connect() as conn:
+                # Database should not exist since create_db() calls drop_db() first
+                conn.execute(text(f'CREATE DATABASE "{target_db}"'))
+        finally:
+            engine.dispose()
+
+    elif scheme == "sqlite":
+        file_path = _get_sqlite_file_path(uri)
+
+        # In-memory databases don't need creation
+        if file_path is None:
+            return
+
+        # For file-based SQLite, ensure the directory exists
+        if file_path.parent:
+            file_path.parent.mkdir(parents=True, exist_ok=True)
+
+        # Create an empty file to "create" the database
+        # SQLite will create the file on first connection, but we'll touch it here
+        if not file_path.exists():
+            file_path.touch()
+
+    else:
+        raise ValueError(f"Unsupported database scheme: {scheme}")
+
+
+def drop_database(uri):
+    """Drop a database.
+
+    Args:
+        uri: Database URI (postgresql://... or sqlite:///...)
+    """
+    parsed = _parse_database_uri(uri)
+    scheme = parsed["scheme"]
+
+    if scheme == "postgresql":
+        admin_uri = _get_postgres_admin_uri(uri)
+        target_db = parsed["database"]
+
+        engine = create_engine(admin_uri, isolation_level="AUTOCOMMIT")
+        try:
+            with engine.connect() as conn:
+                # Terminate any existing connections to the database
+                conn.execute(
+                    text(
+                        """
+                        SELECT pg_terminate_backend(pg_stat_activity.pid)
+                        FROM pg_stat_activity
+                        WHERE pg_stat_activity.datname = :dbname
+                        AND pid <> pg_backend_pid()
+                        """
+                    ),
+                    {"dbname": target_db},
+                )
+                # Drop the database
+                conn.execute(text(f'DROP DATABASE IF EXISTS "{target_db}"'))
+        finally:
+            engine.dispose()
+
+    elif scheme == "sqlite":
+        file_path = _get_sqlite_file_path(uri)
+
+        # In-memory databases don't need dropping
+        if file_path is None:
+            return
+
+        # For file-based SQLite, delete the file
+        if file_path.exists():
+            file_path.unlink()
+
+    else:
+        raise ValueError(f"Unsupported database scheme: {scheme}")