def get_config_from_dict()

in src/graph_notebook/configuration/get_config.py [0:0]

• 59 lines of code
• 24 McCabe index (conditional complexity)

def get_config_from_dict(data: dict, neptune_hosts: list = NEPTUNE_CONFIG_HOST_IDENTIFIERS) -> Configuration:

    if 'ssl_verify' in data:
        ssl_verify = False if data['ssl_verify'] in false_str_variants else True
    else:
        ssl_verify = True
    sparql_section = SparqlSection(**data['sparql']) if 'sparql' in data else SparqlSection('')
    neo4j_section = Neo4JSection(**data['neo4j']) \
        if 'neo4j' in data else Neo4JSection('', '', True, '')
    proxy_host = str(data['proxy_host']) if 'proxy_host' in data else ''
    proxy_port = int(data['proxy_port']) if 'proxy_port' in data else 8182

    is_neptune_host = is_allowed_neptune_host(hostname=data["host"], host_allowlist=neptune_hosts)

    if is_neptune_host:
        if 'neptune_service' in data:
            neptune_service = normalize_service_name(data['neptune_service'])
        else:
            neptune_service = NEPTUNE_DB_SERVICE_NAME
        if 'gremlin' in data:
            if 'connection_protocol' not in data['gremlin']:
                data['gremlin']['connection_protocol'] = DEFAULT_WS_PROTOCOL \
                    if neptune_service == NEPTUNE_DB_SERVICE_NAME else DEFAULT_HTTP_PROTOCOL
            gremlin_section = GremlinSection(**data['gremlin'],
                                             include_protocol=True,
                                             neptune_service=neptune_service)
            if gremlin_section.to_dict()['traversal_source'] != 'g':
                print('Ignoring custom traversal source, Amazon Neptune does not support this functionality.\n')
        else:
            protocol = DEFAULT_WS_PROTOCOL if neptune_service == NEPTUNE_DB_SERVICE_NAME else DEFAULT_HTTP_PROTOCOL
            gremlin_section = GremlinSection(include_protocol=True,
                                             connection_protocol=protocol,
                                             neptune_service=neptune_service)
        if neo4j_section.to_dict()['username'] != DEFAULT_NEO4J_USERNAME \
                or neo4j_section.to_dict()['password'] != DEFAULT_NEO4J_PASSWORD:
            print('Ignoring Neo4J custom authentication, Amazon Neptune does not support this functionality.\n')
        if neo4j_section.to_dict()['database'] != DEFAULT_NEO4J_DATABASE:
            print('Ignoring Neo4J custom database, Amazon Neptune does not support multiple databases.\n')
        config = Configuration(host=data['host'], port=data['port'],
                               neptune_service=neptune_service,
                               auth_mode=AuthModeEnum(data['auth_mode']),
                               ssl=data['ssl'], ssl_verify=ssl_verify, load_from_s3_arn=data['load_from_s3_arn'],
                               aws_region=data['aws_region'], sparql_section=sparql_section,
                               gremlin_section=gremlin_section, neo4j_section=neo4j_section,
                               proxy_host=proxy_host, proxy_port=proxy_port, neptune_hosts=neptune_hosts)
    else:
        excluded_params = []
        for p in neptune_params:
            if p in data:
                excluded_params.append(p)
        if 'gremlin' in data:
            for gp in neptune_gremlin_params:
                if gp in data['gremlin']:
                    excluded_params.append(gp)
        if excluded_params:
            print(f"The provided configuration contains the following parameters that are incompatible with the "
                  f"specified host: {str(excluded_params)}. These parameters have not been saved.\n")
        gremlin_section = GremlinSection(**data['gremlin']) if 'gremlin' in data else GremlinSection()

        config = Configuration(host=data['host'], port=data['port'], ssl=data['ssl'], ssl_verify=ssl_verify,
                               sparql_section=sparql_section, gremlin_section=gremlin_section, neo4j_section=neo4j_section,
                               proxy_host=proxy_host, proxy_port=proxy_port)
    return config