def __init__()

in common/big_query/schema_provider.py [0:0]

• 141 lines of code
• 1 McCabe index (conditional complexity)

    def __init__(self):
        """
        Initializes the SchemaProvider with predefined schemas
        for specific tables.
        """
        self.tables = {
            "tag_templates": {
                "schema": [
                    bigquery.SchemaField(
                        name="resourceName",
                        field_type="STRING",
                        mode="REQUIRED",
                        description=(
                            "Format: projects/:project/locations/"
                            ":location/tagTemplates/:tagTemplateId"
                        ),
                    ),
                    bigquery.SchemaField(
                        name="dataplexResourceName",
                        field_type="STRING",
                        description=(
                            "Format: projects/:project/locations/"
                            "global/aspectTypes/:aspectTypeId"
                        ),
                    ),
                    bigquery.SchemaField(
                        name="projectId", field_type="STRING", mode="REQUIRED"
                    ),
                    bigquery.SchemaField(
                        name="location", field_type="STRING", mode="REQUIRED"
                    ),
                    bigquery.SchemaField(
                        name="tagTemplateId",
                        field_type="STRING",
                        mode="REQUIRED",
                    ),
                    bigquery.SchemaField(
                        name="managingSystem",
                        field_type="STRING",
                        mode="REQUIRED",
                        description="Either DATA_CATALOG or DATAPLEX",
                    ),
                    bigquery.SchemaField(
                        name="isPubliclyReadable", field_type="BOOL"
                    ),
                    bigquery.SchemaField(
                        name="createdAt", field_type="DATE", mode="REQUIRED"
                    ),
                ],
                "is_partitioned": True,
                "partition_column": "createdAt",
                "require_partition_filter": True,
            },
            "entry_groups": {
                "schema": [
                    bigquery.SchemaField(
                        name="resourceName",
                        field_type="STRING",
                        mode="REQUIRED",
                        description=(
                            "Format: projects/:project/locations/"
                            ":location/entryGroups/:entryGroupId"
                        ),
                    ),
                    bigquery.SchemaField(
                        name="dataplexResourceName",
                        field_type="STRING",
                        description=(
                            "Format: projects/:project/locations/"
                            ":location/entryGroups/:entryGroupId"
                        ),
                    ),
                    bigquery.SchemaField(
                        name="projectId", field_type="STRING", mode="REQUIRED"
                    ),
                    bigquery.SchemaField(
                        name="location", field_type="STRING", mode="REQUIRED"
                    ),
                    bigquery.SchemaField(
                        name="entryGroupId",
                        field_type="STRING",
                        mode="REQUIRED",
                    ),
                    bigquery.SchemaField(
                        name="managingSystem",
                        field_type="STRING",
                        mode="REQUIRED",
                        description="Either DATA_CATALOG or DATAPLEX",
                    ),
                    bigquery.SchemaField(
                        name="createdAt", field_type="DATE", mode="REQUIRED"
                    ),
                ],
                "is_partitioned": True,
                "partition_column": "createdAt",
                "require_partition_filter": True,
            },
            "projects": {
                "schema": [
                    bigquery.SchemaField(
                        name="projectId",
                        field_type="STRING",
                        mode="REQUIRED",
                    ),
                    bigquery.SchemaField(
                        name="projectNumber",
                        field_type="INTEGER",
                        mode="REQUIRED",
                    ),
                    bigquery.SchemaField(
                        name="isDataCatalogApiEnabled",
                        field_type="BOOLEAN",
                        mode="REQUIRED",
                    ),
                    bigquery.SchemaField(
                        name="isDataplexApiEnabled",
                        field_type="BOOLEAN",
                        mode="REQUIRED",
                    ),
                    bigquery.SchemaField(
                        name="ancestry",
                        field_type="RECORD",
                        mode="REPEATED",
                        fields=[
                            bigquery.SchemaField(
                                name="type",
                                field_type="STRING",
                                mode="REQUIRED",
                            ),
                            bigquery.SchemaField(
                                name="id",
                                field_type="STRING",
                                mode="REQUIRED",
                            ),
                        ],
                    ),
                    bigquery.SchemaField(
                        name="createdAt", field_type="DATE", mode="REQUIRED"
                    ),
                ],
                "is_partitioned": True,
                "partition_column": "createdAt",
                "require_partition_filter": True,
            },
        }