How to use pybigquery - 10 common examples

def test_only_dataset():
    url = parse_url(make_url('bigquery:///some-dataset'))
    project_id, location, dataset_id, arraysize, credentials_path, job_config = url

    assert project_id is None
    assert location is None
    assert dataset_id == 'some-dataset'
    assert arraysize is None
    assert credentials_path is None
    assert isinstance(job_config, QueryJobConfig)
    # we can't actually test that the dataset is on the job_config,

def test_empty_with_non_config():
    url = parse_url(make_url('bigquery:///?location=some-location&arraysize=1000&credentials_path=/some/path/to.json'))
    project_id, location, dataset_id, arraysize, credentials_path, job_config = url

    assert project_id is None
    assert location == 'some-location'
    assert dataset_id is None
    assert arraysize == 1000
    assert credentials_path == '/some/path/to.json'
    assert job_config is None

def test_not_implemented(not_implemented_arg):
    url = make_url('bigquery://some-project/some-dataset/?' + not_implemented_arg + '=' + 'whatever')
    with pytest.raises(NotImplementedError):
        parse_url(url)

def test_bad_values(param, value):
    url = make_url('bigquery:///?' + param + '=' + value)
    with pytest.raises(ValueError):
        parse_url(url)

def test_empty_url():
    for value in parse_url(make_url('bigquery://')):
        assert value is None

    for value in parse_url(make_url('bigquery:///')):
        assert value is None

def test_basic(url_with_everything):
    project_id, location, dataset_id, arraysize, credentials_path, job_config = parse_url(url_with_everything)

    assert project_id == 'some-project'
    assert location == 'some-location'
    assert dataset_id == 'some-dataset'
    assert arraysize == 1000
    assert credentials_path == '/some/path/to.json'
    assert isinstance(job_config, QueryJobConfig)

def test_disallowed(disallowed_arg):
    url = make_url('bigquery://some-project/some-dataset/?' + disallowed_arg + '=' + 'whatever')
    with pytest.raises(ValueError):
        parse_url(url)

def test_all_values(url_with_everything, param, value):
    job_config = parse_url(url_with_everything)[5]

    config_value = getattr(job_config, param)
    if callable(value):
        assert value(config_value)
    else:
        assert config_value == value

def api_client():
    return ApiClient()

class UniversalSet(object):
    """
    Set containing everything
    https://github.com/dropbox/PyHive/blob/master/pyhive/common.py
    """

    def __contains__(self, item):
        return True


class BigQueryIdentifierPreparer(IdentifierPreparer):
    """
    Set containing everything
    https://github.com/dropbox/PyHive/blob/master/pyhive/sqlalchemy_presto.py
    """
    reserved_words = UniversalSet()

    def __init__(self, dialect):
        super(BigQueryIdentifierPreparer, self).__init__(
            dialect,
            initial_quote="`",
        )

    def quote_column(self, value):
        """
        Quote a column.
        Fields are quoted separately from the record name.
        """

        parts = value.split('.')
        return '.'.join(self.quote_identifier(x) for x in parts)