def METHOD_NAME(self, context):
    # we can't do this in the init because templating occurs in the super init call
    self.source_objects = list(map(self.fix_prefix, self.source_objects))
    # Basically for backwards support of tasks that have nested fields and
    # were created when we were using airflow bigquery hooks.
    # e.g. dags/gtfs_schedule_history/validation_report.yml
    # These tables should be defined using SqlQueryOperator and raw SQL now.
    if self.use_bq_client:
        _bq_client_create_external_table(
            self.destination_project_dataset_table,
            self.schema_fields,
            self.source_objects,
            self.source_format,
            self.geojson,
            self.hive_options,
            self.bucket,
            self.post_hook,
        )
    else:
        if self.hive_options:
            raise RuntimeError(
                "have to use the bigquery client when creating a hive partitioned table"
            )
        field_strings = [
            f'{entry["name"]} {entry["type"]}' for entry in self.schema_fields
        ]
        fields_spec = ",\n".join(field_strings)
        options = [
            f'format = "{self.source_format}"',
            f"uris = {repr(self.source_objects)}",
        ]
        if self.source_format == "CSV":
            options.append(f"skip_leading_rows = {self.skip_leading_rows}")
            options.append(f"field_delimiter = {repr(self.field_delimiter)}")
        if self.geojson:
            options.append("json_extension = 'GEOJSON'")
        options_str = ",".join(options)
        query = f"""