gcp_bigquery_client/
lib.rs

1//! [<img alt="github" src="https://img.shields.io/badge/github-lquerel/gcp_bigquery_client-8da0cb?style=for-the-badge&labelColor=555555&logo=github" height="20">](https://github.com/lquerel/gcp-bigquery-client)
2//! [<img alt="crates.io" src="https://img.shields.io/crates/v/gcp_bigquery_client.svg?style=for-the-badge&color=fc8d62&logo=rust" height="20">](https://crates.io/crates/gcp-bigquery-client)
3//! [<img alt="build status" src="https://img.shields.io/github/workflow/status/lquerel/gcp-bigquery-client/Rust/main?style=for-the-badge" height="20">](https://github.com/lquerel/gcp-bigquery-client/actions?query=branch%3Amain)
4//!
5//! An ergonomic async client library for GCP BigQuery.
6//! * Support for dataset, table, streaming API and query (see [status section](#status) for an exhaustive list of supported API endpoints)
7//! * Support Service Account Key authentication (other OAuth flows will be added later)
8//! * Create tables and rows via builder patterns
9//! * Persist complex Rust structs in structured BigQuery tables
10//! * Async API
11//!
12//! <br>
13//!
14//! Other OAuth flows will be added later.
15//!
16//! For a detailed tutorial on the different ways to use GCP BigQuery Client please check out the [GitHub repository](https://github.com/lquerel/gcp-bigquery-client).
17
18#![allow(clippy::result_large_err)]
19
20#[macro_use]
21extern crate serde;
22extern crate serde_json;
23
24use std::env;
25use std::path::PathBuf;
26use std::sync::Arc;
27
28use client_builder::ClientBuilder;
29use reqwest::Response;
30use serde::Deserialize;
31use storage::StorageApi;
32use yup_oauth2::ServiceAccountKey;
33
34use crate::auth::Authenticator;
35use crate::dataset::DatasetApi;
36use crate::error::BQError;
37use crate::job::JobApi;
38use crate::model_api::ModelApi;
39use crate::project::ProjectApi;
40use crate::routine::RoutineApi;
41use crate::table::TableApi;
42use crate::tabledata::TableDataApi;
43
44/// Since yup_oauth2 structs are used as parameters in public functions there is already semver
45/// coupling, as it is an error if consumer uses different version of yup_oauth than gcp-bigquery-client
46/// Export yup_oauth2 so consumers don't need to carefully keep their dependency versions in sync.
47/// (see https://github.com/lquerel/gcp-bigquery-client/pull/86)
48pub use yup_oauth2;
49
50pub mod auth;
51pub mod client_builder;
52pub mod dataset;
53pub mod error;
54pub mod job;
55pub mod model;
56pub mod model_api;
57pub mod project;
58pub mod routine;
59pub mod storage;
60pub mod table;
61pub mod tabledata;
62
63const BIG_QUERY_V2_URL: &str = "https://bigquery.googleapis.com/bigquery/v2";
64const BIG_QUERY_AUTH_URL: &str = "https://www.googleapis.com/auth/bigquery";
65
66/// An asynchronous BigQuery client.
67#[derive(Clone)]
68pub struct Client {
69    dataset_api: DatasetApi,
70    table_api: TableApi,
71    job_api: JobApi,
72    tabledata_api: TableDataApi,
73    routine_api: RoutineApi,
74    model_api: ModelApi,
75    project_api: ProjectApi,
76    storage_api: StorageApi,
77}
78
79impl Client {
80    pub async fn from_authenticator(auth: Arc<dyn Authenticator>) -> Result<Self, BQError> {
81        Self::from_authenticator_and_client(auth, Some(reqwest::Client::new())).await
82    }
83
84    pub async fn from_authenticator_and_client(
85        auth: Arc<dyn Authenticator>,
86        client: Option<reqwest::Client>,
87    ) -> Result<Self, BQError> {
88        let client = if let Some(client) = client {
89            client
90        } else {
91            reqwest::Client::new()
92        };
93        Ok(Self {
94            dataset_api: DatasetApi::new(client.clone(), Arc::clone(&auth)),
95            table_api: TableApi::new(client.clone(), Arc::clone(&auth)),
96            job_api: JobApi::new(client.clone(), Arc::clone(&auth)),
97            tabledata_api: TableDataApi::new(client.clone(), Arc::clone(&auth)),
98            routine_api: RoutineApi::new(client.clone(), Arc::clone(&auth)),
99            model_api: ModelApi::new(client.clone(), Arc::clone(&auth)),
100            project_api: ProjectApi::new(client, Arc::clone(&auth)),
101            storage_api: StorageApi::new(auth).await?,
102        })
103    }
104
105    /// Constructs a new BigQuery client.
106    /// # Argument
107    /// * `sa_key_file` - A GCP Service Account Key file.
108    pub async fn from_service_account_key_file(sa_key_file: &str) -> Result<Self, BQError> {
109        ClientBuilder::new()
110            .build_from_service_account_key_file(sa_key_file)
111            .await
112    }
113
114    /// Constructs a new BigQuery client from a [`ServiceAccountKey`].
115    /// # Argument
116    /// * `sa_key` - A GCP Service Account Key `yup-oauth2` object.
117    /// * `readonly` - A boolean setting whether the acquired token scope should be readonly.
118    ///
119    /// [`ServiceAccountKey`]: https://docs.rs/yup-oauth2/*/yup_oauth2/struct.ServiceAccountKey.html
120    pub async fn from_service_account_key(sa_key: ServiceAccountKey, readonly: bool) -> Result<Self, BQError> {
121        ClientBuilder::new()
122            .build_from_service_account_key(sa_key, readonly)
123            .await
124    }
125
126    pub async fn with_workload_identity(readonly: bool) -> Result<Self, BQError> {
127        ClientBuilder::new().build_with_workload_identity(readonly).await
128    }
129
130    pub(crate) fn v2_base_url(&mut self, base_url: String) -> &mut Self {
131        self.dataset_api.with_base_url(base_url.clone());
132        self.table_api.with_base_url(base_url.clone());
133        self.job_api.with_base_url(base_url.clone());
134        self.tabledata_api.with_base_url(base_url.clone());
135        self.routine_api.with_base_url(base_url.clone());
136        self.model_api.with_base_url(base_url.clone());
137        self.project_api.with_base_url(base_url.clone());
138        self.storage_api.with_base_url(base_url);
139        self
140    }
141
142    pub async fn from_installed_flow_authenticator<S: AsRef<[u8]>, P: Into<PathBuf>>(
143        secret: S,
144        persistant_file_path: P,
145    ) -> Result<Self, BQError> {
146        ClientBuilder::new()
147            .build_from_installed_flow_authenticator(secret, persistant_file_path)
148            .await
149    }
150
151    pub async fn from_installed_flow_authenticator_from_secret_file<P: Into<PathBuf>>(
152        secret_file: &str,
153        persistant_file_path: P,
154    ) -> Result<Self, BQError> {
155        Self::from_installed_flow_authenticator(
156            tokio::fs::read(secret_file)
157                .await
158                .expect("expecting a valid secret file."),
159            persistant_file_path,
160        )
161        .await
162    }
163
164    pub async fn from_application_default_credentials() -> Result<Self, BQError> {
165        ClientBuilder::new().build_from_application_default_credentials().await
166    }
167
168    pub async fn from_authorized_user_secret(secret: &str) -> Result<Self, BQError> {
169        ClientBuilder::new()
170            .build_from_authorized_user_authenticator(secret)
171            .await
172    }
173
174    /// Returns a dataset API handler.
175    pub fn dataset(&self) -> &DatasetApi {
176        &self.dataset_api
177    }
178
179    /// Returns a table API handler.
180    pub fn table(&self) -> &TableApi {
181        &self.table_api
182    }
183
184    /// Returns a job API handler.
185    pub fn job(&self) -> &JobApi {
186        &self.job_api
187    }
188
189    /// Returns a table data API handler.
190    pub fn tabledata(&self) -> &TableDataApi {
191        &self.tabledata_api
192    }
193
194    /// Returns a routine API handler.
195    pub fn routine(&self) -> &RoutineApi {
196        &self.routine_api
197    }
198
199    /// Returns a model API handler.
200    pub fn model(&self) -> &ModelApi {
201        &self.model_api
202    }
203
204    /// Returns a project API handler.
205    pub fn project(&self) -> &ProjectApi {
206        &self.project_api
207    }
208
209    /// Returns a storage API handler.
210    pub fn storage(&self) -> &StorageApi {
211        &self.storage_api
212    }
213
214    /// Returns a mutable storage API handler.
215    pub fn storage_mut(&mut self) -> &mut StorageApi {
216        &mut self.storage_api
217    }
218}
219
220pub(crate) fn urlencode<T: AsRef<str>>(s: T) -> String {
221    url::form_urlencoded::byte_serialize(s.as_ref().as_bytes()).collect()
222}
223
224async fn process_response<T: for<'de> Deserialize<'de>>(resp: Response) -> Result<T, BQError> {
225    if resp.status().is_success() {
226        Ok(resp.json().await?)
227    } else {
228        Err(BQError::ResponseError {
229            error: resp.json().await?,
230        })
231    }
232}
233
234pub fn env_vars() -> (String, String, String, String) {
235    let project_id = env::var("PROJECT_ID").expect("Environment variable PROJECT_ID");
236    let dataset_id = env::var("DATASET_ID").expect("Environment variable DATASET_ID");
237    let table_id = env::var("TABLE_ID").expect("Environment variable TABLE_ID");
238    let gcp_sa_key =
239        env::var("GOOGLE_APPLICATION_CREDENTIALS").expect("Environment variable GOOGLE_APPLICATION_CREDENTIALS");
240
241    (project_id, dataset_id, table_id, gcp_sa_key)
242}
243
244pub mod google {
245    #![allow(clippy::all)]
246    #[path = "google.api.rs"]
247    pub mod api;
248
249    #[path = ""]
250    pub mod cloud {
251        #[path = ""]
252        pub mod bigquery {
253            #[path = ""]
254            pub mod storage {
255                #![allow(clippy::all)]
256                #[path = "google.cloud.bigquery.storage.v1.rs"]
257                pub mod v1;
258            }
259        }
260    }
261
262    #[path = "google.rpc.rs"]
263    pub mod rpc;
264}
gcp_bigquery_client/lib.rs

gcp_bigquery_client/
lib.rs
