def update()

in bigquery_etl/cli/metadata.py [0:0]

• 65 lines of code
• 18 McCabe index (conditional complexity)

def update(name: str, sql_dir: Optional[str], project_id: Optional[str]) -> None:
    """Update metadata yaml file."""
    table_metadata_files = paths_matching_name_pattern(
        name, sql_dir, project_id=project_id, files=["metadata.yaml"]
    )
    retained_dataset_roles = ConfigLoader.get(
        "deprecation", "retain_dataset_roles", fallback=[]
    )

    # create and populate the dataset metadata yaml file if it does not exist
    for table_metadata_file in table_metadata_files:
        dataset_metadata_path = (
            Path(table_metadata_file).parent.parent / "dataset_metadata.yaml"
        )
        if not dataset_metadata_path.exists():
            continue
        dataset_metadata = DatasetMetadata.from_file(dataset_metadata_path)
        table_metadata = Metadata.from_file(table_metadata_file)

        dataset_metadata_updated = False
        table_metadata_updated = False

        # set dataset metadata default_table_workgroup_access to table_workgroup_access if not set
        if not dataset_metadata.default_table_workgroup_access:
            dataset_metadata.default_table_workgroup_access = (
                dataset_metadata.workgroup_access
            )
            dataset_metadata_updated = True

        if table_metadata.deprecated:
            # set workgroup: [] if table has been tagged as deprecated
            # this overwrites existing workgroups
            table_metadata.workgroup_access = []
            table_metadata_updated = True
            dataset_metadata.workgroup_access = [
                workgroup
                for workgroup in dataset_metadata.workgroup_access
                if workgroup.get("role") in retained_dataset_roles
            ]
            dataset_metadata_updated = True
        else:
            if table_metadata.workgroup_access is None:
                table_metadata.workgroup_access = []

            for (
                default_workgroup_access
            ) in dataset_metadata.default_table_workgroup_access:
                role_exists = False
                for i, table_workgroup_access in enumerate(
                    table_metadata.workgroup_access
                ):
                    if table_workgroup_access.role == default_workgroup_access.get(
                        "role"
                    ):
                        role_exists = True
                        table_metadata.workgroup_access[i].members = sorted(
                            set(table_workgroup_access.members)
                            | set(default_workgroup_access.get("members", []))
                        )
                        table_metadata_updated = True

                if not role_exists:
                    table_metadata.workgroup_access.append(
                        WorkgroupAccessMetadata(
                            role=default_workgroup_access["role"],
                            members=default_workgroup_access.get("members", []),
                        )
                    )
                    table_metadata_updated = True

        if dataset_metadata_updated:
            dataset_metadata.write(dataset_metadata_path)
            click.echo(f"Updated {dataset_metadata_path}")
        if table_metadata_updated:
            table_metadata.write(table_metadata_file)
            click.echo(f"Updated {table_metadata_file}")

    return None