in mysqlx-connector-python/lib/mysqlx/connection.py [0:0]
def __init__(self, settings: Dict[str, Any]) -> None:
self.use_pure: bool = settings.get("use-pure", Protobuf.use_pure)
self._settings: Dict[str, Any] = settings
# Check for DNS SRV
if settings.get("host") and settings.get("dns-srv"):
if not HAVE_DNSPYTHON:
raise InterfaceError(
"MySQL host configuration requested DNS "
"SRV. This requires the Python dnspython "
"module. Please refer to documentation"
)
try:
srv_records = dns.resolver.query(settings["host"], "SRV")
except dns.exception.DNSException as err:
raise InterfaceError(
f"Unable to locate any hosts for '{settings['host']}'"
) from err
self._settings["routers"] = []
for srv in srv_records:
self._settings["routers"].append(
{
"host": srv.target.to_text(omit_final_dot=True),
"port": srv.port,
"priority": srv.priority,
"weight": srv.weight,
}
)
if (
"connection-attributes" not in self._settings
or self._settings["connection-attributes"] is not False
):
self._settings["attributes"] = {}
self._init_attributes()
if "pooling" in settings and settings["pooling"]:
# Create pool and retrieve a Connection instance
PoolsManager().create_pool(settings)
self._connection: Connection = PoolsManager().get_connection(settings)
if self._connection is None:
raise PoolError("Connection could not be retrieved from pool")
else:
self._connection = Connection(self._settings)
self._connection.connect()
# Set default schema
schema = self._settings.get("schema")
if schema:
try:
self.sql(f"USE {quote_identifier(schema)}").execute()
except OperationalError as err:
# Access denied for user will raise err.errno = 1044
errmsg = (
err.msg
if err.errno == 1044
else f"Default schema '{schema}' does not exists"
)
raise InterfaceError(errmsg, err.errno) from err