def __init__()

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