def get_fg_conf()

in repos/features_ingestion_pipeline/infra/feature_store_utils.py [0:0]

• 21 lines of code
• 6 McCabe index (conditional complexity)

def get_fg_conf(file_path: Union[str, Path], bucket_name: str = None) -> dict:
    with open(file_path, "r") as f:
        f_list = json.load(f)
    # offline_conf = None

    # return dict(
    #     event_time_feature_name=f_list["event_time_feature_name"],
    #     record_identifier_feature_name=f_list["record_identifier_feature_name"],
    #     feature_group_name=f_list["feature_group_name"],
    #     online_store_config=f_list["online_store_config"],
    #     offline_store_config=offline_conf,
    #     feature_definitions=prepare_features_definitions(f_list["column_schemas"]),
    # )
    f_list = {pascal2snake(k): v for k, v in f_list.items()}
    f_list["feature_definitions"] = prepare_features_definitions(
        f_list["feature_definitions"]
    )
    if f_list["offline_store_config"]:
        f_list["offline_store_config"] = dict(
            DisableGlueTableCreation=False,
            S3StorageConfig={"S3Uri": f"s3://{bucket_name}/"},
        )
    try:
        f_list["tags"] = [
            CfnTag(**{pascal2snake(j): o for j, o in k.keys()})
            for k in f_list["tags"]
            if k["Key"] != "dev"
        ]
    except:
        pass
    return f_list