launchflow.gcp.bigquery

1
class BigQueryDataset(GCPResource[BigQueryDatasetConnectionInfo])
content_copy

1
from google.cloud import bigquery
2
import launchflow as lf
3

4
# Automatically configures / deploys a BigQuery Dataset in your GCP project
5
dataset = lf.gcp.BigQueryDataset("my-dataset")
6

7
schema = [
8
    bigquery.SchemaField("name", "STRING", mode="REQUIRED"),
9
    bigquery.SchemaField("age", "INTEGER", mode="REQUIRED"),
10
]
11
table = dataset.create_table("table_name", schema=schema)
12

13
dataset.insert_table_data("table_name", [{"name": "Alice", "age": 30}])
14

15
# You can also use the underlying resource directly
16
# For example, for a table with columns name,age
17
query = f"""
18
SELECT name, age
19
FROM `{dataset.dataset_id}.table_name`
20
WHERE age > 10
21
ORDER BY age DESC
22
"""
23

24
for row in dataset.client().query(query):
25
    print(row)
content_copy

1
def __init__(name: str, *, location="US") -> None
content_copy

1
@property
2
def dataset_id() -> str
content_copy

1
def get_table_uuid(table_name: str) -> str
content_copy

1
@lru_cache
2
def client() -> "bigquery.Client"
content_copy

1
@lru_cache
2
def dataset() -> "bigquery.Dataset"
content_copy

1
def create_table(table_name: str,
2
                 *,
3
                 schema: "Optional[List[bigquery.SchemaField]]" = None
4
                 ) -> "bigquery.Table"
content_copy

1
def delete_table(table_name: str) -> None
content_copy

1
def load_table_data_from_csv(table_name: str, file_path: Path) -> None
content_copy

1
def insert_table_data(table_name: str,
2
                      rows_to_insert: List[Dict[Any, Any]]) -> None
content_copy