def split_table_name()

in projects/database-archival/src/database_archival/common/utils/bigquery_utils.py [0:0]

• 27 lines of code
• 12 McCabe index (conditional complexity)

def split_table_name(table_name: str) -> Mapping[str, str]:
    """Splits a table name into a table resource.

    Args:
        table_name: Table name in format <project_id>.<dataset_id>.<table_id>.

    Raises:
        ValueError: no table_name value.
        ValueError: table_name with no dots.
        ValueError: table_name with not enough parts to make project, dataset
            and table components.
        ValueError: table_name with one of project, dataset or table with empty
            values.

    Returns:
        Table resource broken down with project_id, dataset_id and table_id.
    """
    if not table_name:
        raise ValueError(_INVALID_TABLE_MESSAGE + ' Got empty value.')

    if '.' not in table_name:
        raise ValueError(_INVALID_TABLE_MESSAGE + f' Got: {table_name}.')

    if ':' in table_name:
        project_id, rest = table_name.split(':', maxsplit=1)
        if ':' in rest:
            raise ValueError(_INVALID_TABLE_MESSAGE + f' Got: {table_name}.')

        dataset_id, table_id = rest.split('.', maxsplit=1)
        if not project_id or not dataset_id or not table_id:
            raise ValueError(_INVALID_TABLE_MESSAGE + f' Got: {table_name}.')
        return {
            'project_id': project_id,
            'dataset_id': dataset_id,
            'table_id': table_id,
        }

    if len(table_name.split('.')) < 3:
        raise ValueError(_INVALID_TABLE_MESSAGE + f' Got: {table_name}.')

    project_id, dataset_id, table_id = table_name.split('.', maxsplit=2)
    if not project_id or not dataset_id or not table_id:
        raise ValueError(_INVALID_TABLE_MESSAGE + f' Got: {table_name}.')
    return {
        'project_id': project_id,
        'dataset_id': dataset_id,
        'table_id': table_id,
    }