Add support for MSSQL when creating/dropping db

conftest.py

@@ -77,12 +77,38 @@ def sqlite_file_dsn():
     return 'sqlite:///{0}.db'.format(db_name)
 
 
+@pytest.fixture
+def mssql_db_user():
+    return os.environ.get('SQLALCHEMY_UTILS_TEST_MSSQL_USER', 'sa')
+
+
+@pytest.fixture
+def mssql_db_password():
+    return os.environ.get('SQLALCHEMY_UTILS_TEST_MSSQL_PASSWORD',
+                          'Strong!Passw0rd')
+
+
+@pytest.fixture
+def mssql_db_driver():
+    driver = os.environ.get('SQLALCHEMY_UTILS_TEST_MSSQL_DRIVER',
+                            'ODBC Driver 17 for SQL Server')
+    return driver.replace(' ', '+')
+
+
+@pytest.fixture
+def mssql_dsn(mssql_db_user, mssql_db_password, mssql_db_driver, db_name):
+    return 'mssql+pyodbc://{0}:{1}@localhost/{2}?driver={3}'\
+        .format(mssql_db_user, mssql_db_password, db_name, mssql_db_driver)
+
+
 @pytest.fixture
 def dsn(request):
     if 'postgresql_dsn' in request.fixturenames:
         return request.getfuncargvalue('postgresql_dsn')
     elif 'mysql_dsn' in request.fixturenames:
         return request.getfuncargvalue('mysql_dsn')
+    elif 'mssql_dsn' in request.fixturenames:
+        return request.getfuncargvalue('mssql_dsn')
     elif 'sqlite_file_dsn' in request.fixturenames:
         return request.getfuncargvalue('sqlite_file_dsn')
     elif 'sqlite_memory_dsn' in request.fixturenames:

setup.py

@@ -36,6 +36,7 @@ def get_version():
         'pymysql',
         'flake8>=2.4.0',
         'isort>=4.2.2',
+        'pyodbc',
     ],
     'anyjson': ['anyjson>=0.3.3'],
     'babel': ['Babel>=1.3'],

sqlalchemy_utils/functions/database.py

@@ -530,10 +530,15 @@ def create_database(url, encoding='utf8', template=None):
 
     if url.drivername.startswith('postgres'):
         url.database = 'postgres'
+    elif url.drivername.startswith('mssql'):
+        url.database = 'master'
     elif not url.drivername.startswith('sqlite'):
         url.database = None
 
-    engine = sa.create_engine(url)
+    if url.drivername == 'mssql+pyodbc':
+        engine = sa.create_engine(url, connect_args={'autocommit': True})
+    else:
+        engine = sa.create_engine(url)
     result_proxy = None
 
     if engine.dialect.name == 'postgresql':
@@ -592,10 +597,15 @@ def drop_database(url):
 
     if url.drivername.startswith('postgres'):
         url.database = 'postgres'
+    elif url.drivername.startswith('mssql'):
+        url.database = 'master'
     elif not url.drivername.startswith('sqlite'):
         url.database = None
 
-    engine = sa.create_engine(url)
+    if url.drivername == 'mssql+pyodbc':
+        engine = sa.create_engine(url, connect_args={'autocommit': True})
+    else:
+        engine = sa.create_engine(url)
     conn_resource = None
 
     if engine.dialect.name == 'sqlite' and database != ':memory:':

tests/functions/test_database.py

@@ -118,3 +118,12 @@ def test_create_database_twice(self, postgresql_db_user):
         for dsn_item in dsn_list:
             drop_database(dsn_item)
             assert not database_exists(dsn_item)
+
+
+@pytest.mark.usefixtures('mssql_dsn')
+class TestDatabaseMssql(DatabaseTest):
+
+    @pytest.fixture
+    def db_name(self):
+        pytest.importorskip('pyodbc')
+        return 'db_test_sqlalchemy_util'