在 Python 中，如何构造一个 URL 来测试我的 SQL 服务器实例是否为 运行？
In Python, how do I construct a URL to test if my SQL Server instance is running?
我正在使用 Python 3.8 和 pytest-docker-compose 插件 -- https://pypi.org/project/pytest-docker-compose/。有谁知道如何写一个 URL 最终会告诉我我的 SQL 服务器是 运行 吗？
我有这个 docker-compose.yml 文件
version: "3.2"
services:
sql-server-db:
build: ./
container_name: sql-server-db
image: microsoft/mssql-server-linux:2017-latest
ports:
- "1433:1433"
environment:
SA_PASSWORD: "password"
ACCEPT_EULA: "Y"
但我不知道要将什么 URL 传递给我的 Retry 对象以测试服务器是 运行。这失败了...
import pytest
import requests
from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter
...
@pytest.fixture(scope="function")
def wait_for_api(function_scoped_container_getter):
"""Wait for sql server to become responsive"""
request_session = requests.Session()
retries = Retry(total=5,
backoff_factor=0.1,
status_forcelist=[500, 502, 503, 504])
request_session.mount('http://', HTTPAdapter(max_retries=retries))
service = function_scoped_container_getter.get("sql-server-db").network_info[0]
api_url = "http://%s:%s/" % (service.hostname, service.host_port)
assert request_session.get(api_url)
return request_session, api_url
有这个例外
raise ConnectionError(e, request=request)
E requests.exceptions.ConnectionError: HTTPConnectionPool(host='0.0.0.0', port=1433): Max retries exceeded with url: / (Caused by ProtocolError('Connection aborted.', RemoteDisconnected('Remote end closed connection without response')))
if connection.is_connected():
db_Info = connection.get_server_info()
print("Connected to MySQL Server version ", db_Info)
cursor = connection.cursor()
cursor.execute("select database();")
record = cursor.fetchone()
print("You're connected to database: ", record)
你可以使用类似这样的东西，如果它已连接它就会输出
这是一个示例函数，它将重试连接到数据库，并且在成功连接或达到定义的 maxRetries 之前不会 return：
def waitDb(server, database, username, password, maxAttempts, waitBetweenAttemptsSeconds):
"""
Returns True if the connection is successfully established before the maxAttempts number is reached
Conversely returns False
pyodbc.connect has a built-in timeout. Use a waitBetweenAttemptsSeconds greater than zero to add a delay on top of this timeout
"""
for attemptNumber in range(maxAttempts):
cnxn = None
try:
cnxn = pyodbc.connect('DRIVER={ODBC Driver 17 for SQL Server};SERVER='+server+';DATABASE='+database+';UID='+username+';PWD='+ password)
cursor = cnxn.cursor()
except Exception as e:
print(traceback.format_exc())
finally:
if cnxn:
print("The DB is up and running: ")
return True
else:
print("DB not running yet on attempt numer " + str(attemptNumber))
time.sleep(waitBetweenAttemptsSeconds)
print("Max attempts waiting for DB to come online exceeded")
return False
我在这里写了一个最小的例子：https://github.com/claudiumocanu/flow-pytest-docker-compose-wait-mssql。
我包含了三个可以独立执行的动作，但是你可以根据你的要求跳到最后一步：
1.从 python 连接到由撰写文件启动的 mssql：
对我来说，找到并安装合适的 ODBC 驱动程序及其依赖项非常烦人 - ODBC Driver 17 for SQL Server 在 Ubuntu 18.
上对我来说效果最好
要仅执行此步骤，docker-compose up docker-compose.yml in my example, then run the example-connect.py
2。创建了一个尝试使用 maxAttemptsNumber 和重试之间的延迟连接到数据库的函数：
就 运行 这个 example-waitDb.py。您可以使用 maxAttempts 和 delayBetweenAttempts 值，然后随机调出数据库进行测试。
3。将所有内容放在 test_db.py 测试套件中：
- 上述waitDb函数。
- 与您在示例中提供的相同 wrapper and annotations 以启动组合文件中定义的资源
- 一个不会在
waitDb returns 之前执行的虚拟 integration test （如果你想完全阻止这个测试，你可以抛出而不是 returning来自 waitDb 函数的 False)
PS：继续使用 ENVs/vault 等，而不是像我在虚拟示例中那样存储真实密码。
我正在使用 Python 3.8 和 pytest-docker-compose 插件 -- https://pypi.org/project/pytest-docker-compose/。有谁知道如何写一个 URL 最终会告诉我我的 SQL 服务器是 运行 吗？
我有这个 docker-compose.yml 文件
version: "3.2"
services:
sql-server-db:
build: ./
container_name: sql-server-db
image: microsoft/mssql-server-linux:2017-latest
ports:
- "1433:1433"
environment:
SA_PASSWORD: "password"
ACCEPT_EULA: "Y"
但我不知道要将什么 URL 传递给我的 Retry 对象以测试服务器是 运行。这失败了...
import pytest
import requests
from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter
...
@pytest.fixture(scope="function")
def wait_for_api(function_scoped_container_getter):
"""Wait for sql server to become responsive"""
request_session = requests.Session()
retries = Retry(total=5,
backoff_factor=0.1,
status_forcelist=[500, 502, 503, 504])
request_session.mount('http://', HTTPAdapter(max_retries=retries))
service = function_scoped_container_getter.get("sql-server-db").network_info[0]
api_url = "http://%s:%s/" % (service.hostname, service.host_port)
assert request_session.get(api_url)
return request_session, api_url
有这个例外
raise ConnectionError(e, request=request)
E requests.exceptions.ConnectionError: HTTPConnectionPool(host='0.0.0.0', port=1433): Max retries exceeded with url: / (Caused by ProtocolError('Connection aborted.', RemoteDisconnected('Remote end closed connection without response')))
if connection.is_connected():
db_Info = connection.get_server_info()
print("Connected to MySQL Server version ", db_Info)
cursor = connection.cursor()
cursor.execute("select database();")
record = cursor.fetchone()
print("You're connected to database: ", record)
你可以使用类似这样的东西，如果它已连接它就会输出
这是一个示例函数，它将重试连接到数据库，并且在成功连接或达到定义的 maxRetries 之前不会 return：
def waitDb(server, database, username, password, maxAttempts, waitBetweenAttemptsSeconds):
"""
Returns True if the connection is successfully established before the maxAttempts number is reached
Conversely returns False
pyodbc.connect has a built-in timeout. Use a waitBetweenAttemptsSeconds greater than zero to add a delay on top of this timeout
"""
for attemptNumber in range(maxAttempts):
cnxn = None
try:
cnxn = pyodbc.connect('DRIVER={ODBC Driver 17 for SQL Server};SERVER='+server+';DATABASE='+database+';UID='+username+';PWD='+ password)
cursor = cnxn.cursor()
except Exception as e:
print(traceback.format_exc())
finally:
if cnxn:
print("The DB is up and running: ")
return True
else:
print("DB not running yet on attempt numer " + str(attemptNumber))
time.sleep(waitBetweenAttemptsSeconds)
print("Max attempts waiting for DB to come online exceeded")
return False
我在这里写了一个最小的例子：https://github.com/claudiumocanu/flow-pytest-docker-compose-wait-mssql。
我包含了三个可以独立执行的动作，但是你可以根据你的要求跳到最后一步：
1.从 python 连接到由撰写文件启动的 mssql：
对我来说，找到并安装合适的 ODBC 驱动程序及其依赖项非常烦人 - ODBC Driver 17 for SQL Server 在 Ubuntu 18.
上对我来说效果最好
要仅执行此步骤，docker-compose up docker-compose.yml in my example, then run the example-connect.py
2。创建了一个尝试使用 maxAttemptsNumber 和重试之间的延迟连接到数据库的函数：
就 运行 这个 example-waitDb.py。您可以使用 maxAttempts 和 delayBetweenAttempts 值，然后随机调出数据库进行测试。
3。将所有内容放在 test_db.py 测试套件中：
- 上述waitDb函数。
- 与您在示例中提供的相同 wrapper and annotations 以启动组合文件中定义的资源
- 一个不会在
waitDbreturns 之前执行的虚拟 integration test （如果你想完全阻止这个测试，你可以抛出而不是 returning来自waitDb函数的 False)
PS：继续使用 ENVs/vault 等，而不是像我在虚拟示例中那样存储真实密码。