def _sql_meta()

in aws_xray_sdk/ext/sqlalchemy_core/patch.py [0:0]

• 36 lines of code
• 2 McCabe index (conditional complexity)

def _sql_meta(engine_instance, args):
    try:
        metadata = {}
        # Workaround for https://github.com/sqlalchemy/sqlalchemy/issues/10662
        # sqlalchemy.engine.url.URL's __repr__ does not url encode username nor password.
        # This will continue to work once sqlalchemy fixes the bug.
        sa_url = engine_instance.engine.url
        username = sa_url.username
        sa_url = sa_url._replace(username=None, password=None)
        url = urlparse(str(sa_url))
        name = url.netloc
        if username:
            # Restore url encoded username
            quoted_username = quote_plus(username)
            url = url._replace(netloc='{}@{}'.format(quoted_username, url.netloc))
        # Add Scheme to uses_netloc or // will be missing from url.
        uses_netloc.append(url.scheme)
        metadata['url'] = url.geturl()
        metadata['user'] = url.username
        metadata['database_type'] = engine_instance.engine.name
        try:
            version = getattr(engine_instance.dialect, '{}_version'.format(engine_instance.engine.driver))
            version_str = '.'.join(map(str, version))
            metadata['driver_version'] = "{}-{}".format(engine_instance.engine.driver, version_str)
        except AttributeError:
            metadata['driver_version'] = engine_instance.engine.driver
        if engine_instance.dialect.server_version_info is not None:
            metadata['database_version'] = '.'.join(map(str, engine_instance.dialect.server_version_info))
        if xray_recorder.stream_sql:
            try:
                if isinstance(args[0], ClauseElement):
                    metadata['sanitized_query'] = str(args[0].compile(engine_instance.engine))
                else:
                    metadata['sanitized_query'] = str(args[0])
            except Exception:
                logging.getLogger(__name__).exception('Error getting the sanitized query')
    except Exception:
        metadata = None
        name = None
        logging.getLogger(__name__).exception('Error parsing sql metadata.')
    return name, metadata