forked from distribution/distribution
-
Notifications
You must be signed in to change notification settings - Fork 0
/
bigquery.go
147 lines (126 loc) · 4.14 KB
/
bigquery.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
// Copyright 2015 Google Inc. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package bigquery
// TODO(mcgreevy): support dry-run mode when creating jobs.
import (
"fmt"
"net/http"
"golang.org/x/net/context"
bq "google.golang.org/api/bigquery/v2"
)
// A Source is a source of data for the Copy function.
type Source interface {
implementsSource()
}
// A Destination is a destination of data for the Copy function.
type Destination interface {
implementsDestination()
}
// An Option is an optional argument to Copy.
type Option interface {
implementsOption()
}
// A ReadSource is a source of data for the Read function.
type ReadSource interface {
implementsReadSource()
}
// A ReadOption is an optional argument to Read.
type ReadOption interface {
customizeRead(conf *pagingConf)
}
const Scope = "https://www.googleapis.com/auth/bigquery"
// Client may be used to perform BigQuery operations.
type Client struct {
service service
projectID string
}
// Note: many of the methods on *Client appear in the various *_op.go source files.
// NewClient constructs a new Client which can perform BigQuery operations.
// Operations performed via the client are billed to the specified GCP project.
// The supplied http.Client is used for making requests to the BigQuery server and must be capable of
// authenticating requests with Scope.
func NewClient(client *http.Client, projectID string) (*Client, error) {
bqService, err := newBigqueryService(client)
if err != nil {
return nil, fmt.Errorf("constructing bigquery client: %v", err)
}
c := &Client{
service: bqService,
projectID: projectID,
}
return c, nil
}
// initJobProto creates and returns a bigquery Job proto.
// The proto is customized using any jobOptions in options.
// The list of Options is returned with the jobOptions removed.
func initJobProto(projectID string, options []Option) (*bq.Job, []Option) {
job := &bq.Job{}
var other []Option
for _, opt := range options {
if o, ok := opt.(jobOption); ok {
o.customizeJob(job, projectID)
} else {
other = append(other, opt)
}
}
return job, other
}
// Copy starts a BigQuery operation to copy data from a Source to a Destination.
func (c *Client) Copy(ctx context.Context, dst Destination, src Source, options ...Option) (*Job, error) {
switch dst := dst.(type) {
case *Table:
switch src := src.(type) {
case *GCSReference:
return c.load(ctx, dst, src, options)
case *Table:
return c.cp(ctx, dst, Tables{src}, options)
case Tables:
return c.cp(ctx, dst, src, options)
case *Query:
return c.query(ctx, dst, src, options)
}
case *GCSReference:
if src, ok := src.(*Table); ok {
return c.extract(ctx, dst, src, options)
}
}
return nil, fmt.Errorf("no Copy operation matches dst/src pair: dst: %T ; src: %T", dst, src)
}
// Read fetches data from a ReadSource and returns the data via an Iterator.
func (c *Client) Read(ctx context.Context, src ReadSource, options ...ReadOption) (*Iterator, error) {
switch src := src.(type) {
case *Job:
return c.readQueryResults(src, options)
case *Query:
return c.executeQuery(ctx, src, options...)
case *Table:
return c.readTable(src, options)
}
return nil, fmt.Errorf("src (%T) does not support the Read operation", src)
}
// executeQuery submits a query for execution and returns the results via an Iterator.
func (c *Client) executeQuery(ctx context.Context, q *Query, options ...ReadOption) (*Iterator, error) {
dest := &Table{}
job, err := c.Copy(ctx, dest, q, WriteTruncate)
if err != nil {
return nil, err
}
return c.Read(ctx, job, options...)
}
func (c *Client) Dataset(id string) *Dataset {
return &Dataset{
id: id,
client: c,
}
}