Skip to content

Allow to use dialect mssql+pyodbc #298

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 7 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions mssql/setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,9 @@
install_requires=[
"testcontainers-core",
"sqlalchemy",
# TODO: convert these to extras
"pymssql",
"pyodbc",
],
python_requires=">=3.7",
)
35 changes: 32 additions & 3 deletions mssql/testcontainers/mssql/__init__.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
from os import environ
from typing import Optional
from typing import Optional, Literal
from testcontainers.core.generic import DbContainer


Expand All @@ -21,7 +21,22 @@ class SqlServerContainer(DbContainer):

def __init__(self, image: str = "mcr.microsoft.com/mssql/server:2019-latest", user: str = "SA",
password: Optional[str] = None, port: int = 1433, dbname: str = "tempdb",
dialect: str = 'mssql+pymssql', **kwargs) -> None:
dialect: Literal['mssql+pymssql', 'mssql+pyodbc'] = 'mssql+pymssql', **kwargs) -> None:
"""
Initialize SqlServerContainer

Args:
image: MSSQL Server image. For example, use a specific version
user: DB user name
password: DB password
port: Port to be exposed
dbname: Database name
dialect: SQLAlchemy database dialect. Allowed values are
* 'mssql+pymssql': Uses `pymssql <https://github.com/pymssql/pymssql>`_ driver
* 'mssql+pyodbc': Uses `pyodbc <https://github.com/mkleehammer/pyodbc>`_ driver
This also defines the driver that is used to connect to the database.
kwargs: Keyword arguments passed to initialization of underlying docker container
"""
super(SqlServerContainer, self).__init__(image, **kwargs)

self.port_to_expose = port
Expand All @@ -39,7 +54,21 @@ def _configure(self) -> None:
self.with_env("ACCEPT_EULA", 'Y')

def get_connection_url(self) -> str:
return super()._create_connection_url(
url = super()._create_connection_url(
dialect=self.dialect, username=self.SQLSERVER_USER, password=self.SQLSERVER_PASSWORD,
db_name=self.SQLSERVER_DBNAME, port=self.port_to_expose
)
if self.dialect == "mssql+pyodbc":
import pyodbc
import re
r = re.compile('ODBC Driver \d{1,2} for SQL Server')
# We sort drivers in reversed order to get the latest
drivers = sorted(list(filter(r.match, pyodbc.drivers())), reverse=True)

if len(drivers) > 0:
driver_str = drivers[0].replace(" ", "+")
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

My hunch is that this won't work because sorted is an iterable that can't be indexed. We likely also need to parse the version as an integer because, using string comparison, version 8 is greater than version 10. E.g.,

>>> max(['ODBC Driver 10 for SQL Server', 'ODBC Driver 8 for SQL Server'])
'ODBC Driver 8 for SQL Server'

else:
raise ImportError(f"No driver available for using dialect {self.dialect}")
url += f"?driver={driver_str}"
return url

16 changes: 10 additions & 6 deletions mssql/tests/test_mssql.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,12 +4,16 @@

def test_docker_run_mssql():
image = 'mcr.microsoft.com/azure-sql-edge'
dialect = 'mssql+pymssql'
with SqlServerContainer(image, dialect=dialect) as mssql:
e = sqlalchemy.create_engine(mssql.get_connection_url())
result = e.execute('select @@servicename')
for row in result:
assert row[0] == 'MSSQLSERVER'
dialects = ['mssql+pymssql', 'mssql+pyodbc']
ends_withs = ["tempdb", "for+SQL+Server"]
for dialect, end_with in zip(dialects, ends_withs):
with SqlServerContainer(dialect=dialect) as mssql:
url = mssql.get_connection_url()
assert url.endswith(end_with)
e = sqlalchemy.create_engine(url)
result = e.execute('select @@servicename')
for row in result:
assert row[0] == 'MSSQLSERVER'

with SqlServerContainer(image, password="1Secure*Password2", dialect=dialect) as mssql:
e = sqlalchemy.create_engine(mssql.get_connection_url())
Expand Down