def connect()

in google/cloud/sql/connector/pytds.py [0:0]

• 26 lines of code
• 6 McCabe index (conditional complexity)

def connect(ip_address: str, sock: ssl.SSLSocket, **kwargs: Any) -> "pytds.Connection":
    """Helper function to create a pytds DB-API connection object.

    Args:
        ip_address (str): A string containing an IP address for the Cloud SQL
            instance.
        sock (ssl.SSLSocket): An SSLSocket object created from the Cloud SQL
            server CA cert and ephemeral cert.

    Returns:
        pytds.Connection: A pytds connection to the Cloud SQL
            instance.

    Raises:
        ImportError: The pytds module cannot be imported.
    """
    try:
        import pytds
    except ImportError:
        raise ImportError(
            'Unable to import module "pytds." Please install and try again.'
        )

    db = kwargs.pop("db", None)

    if kwargs.pop("active_directory_auth", False):
        if platform.system() == "Windows":
            # Ignore username and password if using active directory auth
            server_name = kwargs.pop("server_name")
            return pytds.connect(
                database=db,
                auth=pytds.login.SspiAuth(port=1433, server_name=server_name),
                sock=sock,
                **kwargs,
            )
        else:
            raise PlatformNotSupportedError(
                "Active Directory authentication is currently only supported on Windows."
            )

    user = kwargs.pop("user")
    passwd = kwargs.pop("password")
    return pytds.connect(
        ip_address, database=db, user=user, password=passwd, sock=sock, **kwargs
    )