fromprefect_dbt.cli.configsimportBigQueryTargetConfigsfromprefect_gcp.credentialsimportGcpCredentialscredentials=GcpCredentials.load("BLOCK-NAME-PLACEHOLDER")target_configs=BigQueryTargetConfigs(schema="schema",# also known as datasetcredentials=credentials,)
Source code in prefect_dbt/cli/configs/bigquery.py
classBigQueryTargetConfigs(BaseTargetConfigs):""" Target configs contain credentials and settings, specific to BigQuery. To find valid keys, head to the [BigQuery Profile]( https://docs.getdbt.com/reference/warehouse-profiles/bigquery-profile) page. Attributes: credentials: The credentials to use to authenticate; if there are duplicate keys between credentials and TargetConfigs, e.g. schema, an error will be raised. Examples: Load stored BigQueryTargetConfigs. ```python from prefect_dbt.cli.configs import BigQueryTargetConfigs bigquery_target_configs = BigQueryTargetConfigs.load("BLOCK_NAME") ``` Instantiate BigQueryTargetConfigs. ```python from prefect_dbt.cli.configs import BigQueryTargetConfigs from prefect_gcp.credentials import GcpCredentials credentials = GcpCredentials.load("BLOCK-NAME-PLACEHOLDER") target_configs = BigQueryTargetConfigs( schema="schema", # also known as dataset credentials=credentials, ) ``` """_block_type_name="dbt CLI BigQuery Target Configs"_logo_url="https://images.ctfassets.net/gm98wzqotmnx/5zE9lxfzBHjw3tnEup4wWL/9a001902ed43a84c6c96d23b24622e19/dbt-bit_tm.png?h=250"# noqa_description="dbt CLI target configs containing credentials and settings, specific to BigQuery."# noqa_documentation_url="https://prefecthq.github.io/prefect-dbt/cli/configs/bigquery/#prefect_dbt.cli.configs.bigquery.BigQueryTargetConfigs"# noqatype:Literal["bigquery"]=Field(default="bigquery",description="The type of target.")project:Optional[str]=Field(default=None,description="The project to use.")credentials:GcpCredentials=Field(default_factory=GcpCredentials,description="The credentials to use to authenticate.",)defget_configs(self)->Dict[str,Any]:""" Returns the dbt configs specific to BigQuery profile. Returns: A configs JSON. """# since GcpCredentials will always define a projectself_copy=self.copy()ifself_copy.projectisnotNone:self_copy.credentials.project=Noneall_configs_json=self._populate_configs_json({},self_copy.__fields__,model=self_copy)# decouple prefect-gcp from prefect-dbt# by mapping all the keys dbt gcp accepts# https://docs.getdbt.com/reference/warehouse-setups/bigquery-setuprename_keys={# dbt"type":"type","schema":"schema","threads":"threads",# general"dataset":"schema","method":"method","project":"project",# service-account"service_account_file":"keyfile",# service-account json"service_account_info":"keyfile_json",# oauth secrets"refresh_token":"refresh_token","client_id":"client_id","client_secret":"client_secret","token_uri":"token_uri",# optional"priority":"priority","timeout_seconds":"timeout_seconds","location":"location","maximum_bytes_billed":"maximum_bytes_billed","scopes":"scopes","impersonate_service_account":"impersonate_service_account","execution_project":"execution_project",}configs_json={}extras=self.extrasor{}forkeyinall_configs_json.keys():ifkeynotinrename_keysandkeynotinextras:# skip invalid keyscontinue# rename key to something dbt profile expectsdbt_key=rename_keys.get(key)orkeyconfigs_json[dbt_key]=all_configs_json[key]if"keyfile_json"inconfigs_json:configs_json["method"]="service-account-json"elif"keyfile"inconfigs_json:configs_json["method"]="service-account"configs_json["keyfile"]=str(configs_json["keyfile"])else:configs_json["method"]="oauth-secrets"# through gcloud application-default logingoogle_credentials=(self_copy.credentials.get_credentials_from_service_account())ifhasattr(google_credentials,"token"):request=Request()google_credentials.refresh(request)configs_json["token"]=google_credentials.tokenelse:forkeyin("refresh_token","client_id","client_secret","token_uri"):configs_json[key]=getattr(google_credentials,key)if"project"notinconfigs_json:raiseValueError("The keyword, project, must be provided in either ""GcpCredentials or BigQueryTargetConfigs")returnconfigs_json
defget_configs(self)->Dict[str,Any]:""" Returns the dbt configs specific to BigQuery profile. Returns: A configs JSON. """# since GcpCredentials will always define a projectself_copy=self.copy()ifself_copy.projectisnotNone:self_copy.credentials.project=Noneall_configs_json=self._populate_configs_json({},self_copy.__fields__,model=self_copy)# decouple prefect-gcp from prefect-dbt# by mapping all the keys dbt gcp accepts# https://docs.getdbt.com/reference/warehouse-setups/bigquery-setuprename_keys={# dbt"type":"type","schema":"schema","threads":"threads",# general"dataset":"schema","method":"method","project":"project",# service-account"service_account_file":"keyfile",# service-account json"service_account_info":"keyfile_json",# oauth secrets"refresh_token":"refresh_token","client_id":"client_id","client_secret":"client_secret","token_uri":"token_uri",# optional"priority":"priority","timeout_seconds":"timeout_seconds","location":"location","maximum_bytes_billed":"maximum_bytes_billed","scopes":"scopes","impersonate_service_account":"impersonate_service_account","execution_project":"execution_project",}configs_json={}extras=self.extrasor{}forkeyinall_configs_json.keys():ifkeynotinrename_keysandkeynotinextras:# skip invalid keyscontinue# rename key to something dbt profile expectsdbt_key=rename_keys.get(key)orkeyconfigs_json[dbt_key]=all_configs_json[key]if"keyfile_json"inconfigs_json:configs_json["method"]="service-account-json"elif"keyfile"inconfigs_json:configs_json["method"]="service-account"configs_json["keyfile"]=str(configs_json["keyfile"])else:configs_json["method"]="oauth-secrets"# through gcloud application-default logingoogle_credentials=(self_copy.credentials.get_credentials_from_service_account())ifhasattr(google_credentials,"token"):request=Request()google_credentials.refresh(request)configs_json["token"]=google_credentials.tokenelse:forkeyin("refresh_token","client_id","client_secret","token_uri"):configs_json[key]=getattr(google_credentials,key)if"project"notinconfigs_json:raiseValueError("The keyword, project, must be provided in either ""GcpCredentials or BigQueryTargetConfigs")returnconfigs_json