How to use the dbt.adapters.factory.get_adapter function in dbt

def built_schema(project_dir, schema, profiles_dir, test_kwargs, project_def):
    # make our args, write our project out
    args = TestArgs(profiles_dir=profiles_dir, kwargs=test_kwargs)
    project_def.write_to(project_dir)
    # build a config of our own
    os.chdir(project_dir)
    start = os.getcwd()
    try:
        cfg = RuntimeConfig.from_args(args)
    finally:
        os.chdir(start)
    register_adapter(cfg)
    adapter = get_adapter(cfg)
    execute(adapter, 'drop schema if exists {} cascade'.format(schema))
    execute(adapter, 'create schema {}'.format(schema))
    yield
    adapter = get_adapter(cfg)
    adapter.cleanup_connections()
    execute(adapter, 'drop schema if exists {} cascade'.format(schema))

if get_materialization(model) not in ('table', 'incremental'):
        return ''

    sort_keys = model_config.get('sort')
    sort_type = model_config.get('sort_type', 'compound')

    if not isinstance(sort_type, basestring):
        compiler_error(
            model,
            "The provided sort_type '{}' is not valid!".format(sort_type)
        )

    sort_type = sort_type.strip().lower()

    adapter = get_adapter(project.run_environment())
    return adapter.sort_qualifier(sort_type, sort_keys)

def attempt_connection(profile):
        """Return a string containing the error message, or None if there was
        no error.
        """
        register_adapter(profile)
        adapter = get_adapter(profile)
        try:
            with adapter.connection_named('debug'):
                adapter.execute('select 1 as id')
        except Exception as exc:
            return COULD_NOT_CONNECT_MESSAGE.format(
                err=str(exc),
                url=ProfileConfigDocs,
            )

        return None

def get_invocation_context(user, config, args):
    # put this in here to avoid an import cycle
    from dbt.adapters.factory import get_adapter
    try:
        adapter_type = get_adapter(config).type()
    except Exception:
        adapter_type = None

    return {
        "project_id": None if config is None else config.hashed_name(),
        "user_id": user.id,
        "invocation_id": user.invocation_id,

        "command": args.which,
        "options": None,
        "version": str(dbt_version.installed),

        "run_type": get_run_type(args),
        "adapter_type": adapter_type,
    }

def generate_base(model, model_dict, config, manifest, source_config,
                  provider, adapter=None):
    """Generate the common aspects of the config dict."""
    if provider is None:
        raise dbt.exceptions.InternalException(
            "Invalid provider given to context: {}".format(provider))

    target_name = config.target_name
    target = config.to_profile_info()
    del target['credentials']
    target.update(config.credentials.serialize(with_aliases=True))
    target['type'] = config.credentials.type
    target.pop('pass', None)
    target['name'] = target_name

    adapter = get_adapter(config)

    context = {'env': target}

    pre_hooks = None
    post_hooks = None

    db_wrapper = DatabaseWrapper(adapter)

    context = dbt.utils.merge(context, {
        "adapter": db_wrapper,
        "api": {
            "Relation": db_wrapper.Relation,
            "Column": adapter.Column,
        },
        "column": adapter.Column,
        "config": provider.Config(model_dict, source_config),

def generate(model, runtime_config, manifest, source_config):
    # during parsing, we don't have a connection, but we might need one, so we
    # have to acquire it.
    # In the future, it would be nice to lazily open the connection, as in some
    # projects it would be possible to parse without connecting to the db
    with get_adapter(runtime_config).connection_named(model.get('name')):
        return dbt.context.common.generate(
            model, runtime_config, manifest, source_config, Provider()
        )

def run(self):

        # Look up all of the relations in the DB
        adapter = dbt.adapters.factory.get_adapter(self.config)
        manifest = self._get_manifest()

        schemas = set()
        model_relations = set()
        # Look up all of the relations dbt knows about
        for node in manifest.nodes.values():
            if node["resource_type"] != "source":
                schema_info = (node["database"], node["schema"])
                schemas.update([schema_info])
                node = node.to_dict()
                is_refable = (
                    node["resource_type"] in NodeType.refable()
                    or node["resource_type"] == "archive"
                )
                is_enabled = check_is_enabled(node)
                is_ephemeral = node["config"]["materialized"] == "ephemeral"

text = "Concurrency: {} threads (target='{}')"
        concurrency_line = text.format(num_threads, target_name)
        with NodeCount(self.num_nodes):
            dbt.ui.printer.print_timestamped_line(concurrency_line)
        with TextOnly():
            dbt.ui.printer.print_timestamped_line("")

        pool = ThreadPool(num_threads)
        try:
            self.run_queue(pool)

        except KeyboardInterrupt:
            pool.close()
            pool.terminate()

            adapter = get_adapter(self.config)

            if not adapter.is_cancelable():
                msg = ("The {} adapter does not support query "
                       "cancellation. Some queries may still be "
                       "running!".format(adapter.type()))

                yellow = dbt.ui.printer.COLOR_FG_YELLOW
                dbt.ui.printer.print_timestamped_line(msg, yellow)
                raise

            for conn_name in adapter.cancel_open_connections():
                dbt.ui.printer.print_cancel_line(conn_name)

            pool.join()

            dbt.ui.printer.print_run_end_messages(self.node_results,