Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -10,13 +10,15 @@ authors = [
version="0.12.6"
description="Semantic Link Labs for Microsoft Fabric"
readme="README.md"
requires-python=">=3.10,<3.12"
requires-python=">=3.10,<3.13"
classifiers = [
"Development Status :: 3 - Alpha",
"Intended Audience :: Developers",
"Intended Audience :: Education",
"Intended Audience :: Science/Research",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3 :: Only",
"Framework :: Jupyter"
]
Expand Down
6 changes: 6 additions & 0 deletions src/sempy_labs/variable_library/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,9 @@
get_variable_library_definition,
get_variable_values,
get_variable_value,
create_variable_library,
update_variable_library,
update_variable_library_definition,
)

__all__ = [
Expand All @@ -16,4 +19,7 @@
"get_variable_library_definition",
"get_variable_values",
"get_variable_value",
"create_variable_library",
"update_variable_library",
"update_variable_library_definition",
]
349 changes: 348 additions & 1 deletion src/sempy_labs/variable_library/_functions.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,10 +8,11 @@
_decode_b64,
)
import pandas as pd
from typing import Any, Optional, List, Union
from typing import Any, Optional, List, Union, Dict
from uuid import UUID
from sempy._utils._log import log
import json
import base64
import sempy_labs._icons as icons


Expand Down Expand Up @@ -401,3 +402,349 @@ def get_variable_value(
workspace=workspace,
value_set=value_set,
)[variable_name]


def _encode_b64(content: str) -> str:
"""
Encode a string to base64.

Parameters
----------
content : str
The string content to encode.

Returns
-------
str
The base64 encoded string.
"""
return base64.b64encode(content.encode("utf-8")).decode("utf-8")


def _create_variable_library_definition(
variables: Optional[List[Dict]] = None,
value_sets: Optional[List[Dict]] = None,
value_sets_order: Optional[List[str]] = None,
) -> Dict:
"""
Create the definition structure for a variable library.

Parameters
----------
variables : List[Dict], optional
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

instead of 'optional', use default=None. You can see this is the standard across semantic link labs.

List of variable dictionaries with keys: name, type, value, note (optional).
value_sets : List[Dict], optional
List of value set dictionaries with keys: name, variableOverrides.
active_value_set : str, optional
Name of the active value set. Defaults to "Default value set".

Returns
-------
Dict
The definition structure for the API.
"""
parts = []
json_schema_root = "https://developer.microsoft.com/json-schemas/fabric/item/variableLibrary/definition/"

# Default variables if none provided
if variables is None:
variables = []

# Create variables.json part
variables_content = {
"$schema": json_schema_root + "variables/1.0.0/schema.json",
"variables": variables,
}
variables_json = json.dumps(variables_content, separators=(",", ":"))
parts.append(
{
"path": "variables.json",
"payload": _encode_b64(variables_json),
"payloadType": "InlineBase64",
}
)

if value_sets:
for value_set in value_sets:
value_set_name = value_set.get("name")
value_set_data = value_set

# Ensure proper structure for value set
value_set_content = {
"$schema": json_schema_root + "valueSet/1.0.0/schema.json",
"name": value_set_name,
"variableOverrides": value_set_data.get("variableOverrides", []),
}

value_set_json = json.dumps(value_set_content, separators=(",", ":"))
parts.append(
{
"path": f"valueSets/{value_set_name}.json",
"payload": _encode_b64(value_set_json),
"payloadType": "InlineBase64",
}
)

# Create settings.json part
if value_sets_order:
settings_content = {
"$schema": json_schema_root + "settings/1.0.0/schema.json",
"valueSetsOrder": value_sets_order,
}
settings_json = json.dumps(settings_content, separators=(",", ":"))
parts.append(
{
"path": "settings.json",
"payload": _encode_b64(settings_json),
"payloadType": "InlineBase64",
}
)

return {"format": "VariableLibraryV1", "parts": parts}


@log
def create_variable_library(
variable_library_name: str,
description: Optional[str] = None,
variables: Optional[List[Dict]] = None,
value_sets: Optional[Dict[str, Dict]] = None,
value_sets_order: Optional[List[str]] = None,
workspace: Optional[str | UUID] = None,
) -> str:
"""
Creates a new variable library with optional variables and value sets.

This is a wrapper function for the following API: `Items - Create Variable Library <https://learn.microsoft.com/rest/api/fabric/variablelibrary/items/create-variable-library>`_.

Service Principal Authentication is supported (see `here <https://github.com/microsoft/semantic-link-labs/blob/main/notebooks/Service%20Principal.ipynb>`_ for examples).

Parameters
----------
variable_library_name : str
Name of the variable library.
description : str, optional
Description of the variable library.
variables : List[Dict], optional
List of variable dictionaries. Each dict should contain:
- name (str): Variable name
- type (str): Variable type (e.g., "String", "Number", "Boolean")
- value (Any): Variable value
- note (str, optional): Optional note/description for the variable
Example: [{"name": "var1", "type": "String", "value": "test", "note": "A test variable"}]
value_sets : Dict[str, Dict], optional
Dictionary of value sets where key is value set name and value contains:
- variableOverrides (List[Dict]): List of variable overrides
Example: [{"name": "ProductionSet", "variableOverrides": [{"name": "var1", "value": "prod_value"}]}]
value_sets_order : List[str], optional
Order of value sets. If not provided, defaults to [""].
If value_sets are provided and this is None, the first value set will be used.
workspace : str | uuid.UUID, optional
The Fabric workspace name or ID.
Defaults to None which resolves to the workspace of the attached lakehouse
or if no lakehouse attached, resolves to the workspace of the notebook.

Returns
-------
str
Status message indicating the result of the creation operation.
"""

workspace_id = resolve_workspace_id(workspace)

# Create the definition
definition = _create_variable_library_definition(
variables=variables, value_sets=value_sets, value_sets_order=value_sets_order
)

payload = {
"displayName": variable_library_name,
"description": description or "",
"definition": definition,
}

response = _base_api(
request=f"/v1/workspaces/{workspace_id}/variableLibraries",
method="post",
client="fabric_sp",
payload=payload,
status_codes=[201, 202],
lro_return_json=False,
)

if response.status_code == 201:
print(
f"{icons.green_dot} Variable library '{variable_library_name}' created successfully."
)
return "Created"
elif response.status_code == 202:
print(
f"{icons.in_progress} Variable library '{variable_library_name}' creation is in progress."
)
return "In Progress"
else:
print(
f"{icons.red_dot} Failed to create variable library '{variable_library_name}'. Status code: {response.status_code}"
)
result = response.json()
return f"Failed with status code {result.get('errorCode')} and message: {result.get('message')}"


@log
def update_variable_library(
variable_library: str | UUID,
variable_library_name: Optional[str] = None,
description: Optional[str] = None,
active_value_set: Optional[str] = None,
workspace: Optional[str | UUID] = None,
) -> str:
"""
Updates an existing variable library's properties.

This is a wrapper function for the following API: `Items - Update Variable Library <https://learn.microsoft.com/rest/api/fabric/variablelibrary/items/update-variable-library>`_.

Service Principal Authentication is supported (see `here <https://github.com/microsoft/semantic-link-labs/blob/main/notebooks/Service%20Principal.ipynb>`_ for examples).

Parameters
----------
variable_library : str | uuid.UUID
Name or ID of the variable library to update.
variable_library_name : str, optional
New name for the variable library.
description : str, optional
New description for the variable library.
active_value_set : str, optional
Name of the active value set to set for the variable library.
workspace : str | uuid.UUID, optional
The Fabric workspace name or ID.
Defaults to None which resolves to the workspace of the attached lakehouse
or if no lakehouse attached, resolves to the workspace of the notebook.

Returns
-------
str
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

no point in returning a string

Status message indicating the result of the update operation.
"""

workspace_id = resolve_workspace_id(workspace)
variable_library_id = resolve_item_id(
item=variable_library, type="VariableLibrary", workspace=workspace
)

payload = {}

if variable_library_name is not None:
payload["displayName"] = variable_library_name

if description is not None:
payload["description"] = description

if active_value_set is not None:
payload["properties"] = {"activeValueSetName": active_value_set}

if not payload:
print(f"{icons.yellow_dot} No updates provided for variable library.")
return "No updates"

response = _base_api(
request=f"/v1/workspaces/{workspace_id}/VariableLibraries/{variable_library_id}",
method="patch",
client="fabric_sp",
payload=payload,
status_codes=[200],
)

if response.status_code == 200:
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

please follow the convention for other such functions. no need for an if statement here. just print that the variable library was updated successfully. if an issue occurred, an error would be caught by _base_api.

print(f"{icons.green_dot} Variable library updated successfully.")
return "Updated"
else:
print(
f"{icons.red_dot} Failed to update variable library. Status code: {response.status_code}"
)
result = response.json()
return f"Failed with status code {result.get('errorCode')} and message: {result.get('message')}"


@log
def update_variable_library_definition(
variable_library: str | UUID,
variables: Optional[List[Dict]] = None,
value_sets: Optional[Dict[str, Dict]] = None,
value_sets_order: Optional[List[str]] = None,
workspace: Optional[str | UUID] = None,
) -> str:
"""
Updates the definition of an existing variable library.

This is a wrapper function for the following API: `Items - Update Variable Library Definition <https://learn.microsoft.com/rest/api/fabric/variablelibrary/items/update-variable-library-definition>`_.

Service Principal Authentication is supported (see `here <https://github.com/microsoft/semantic-link-labs/blob/main/notebooks/Service%20Principal.ipynb>`_ for examples).

Parameters
----------
variable_library : str | uuid.UUID
Name or ID of the variable library to update.
variables : List[Dict], optional
List of variable dictionaries. Each dict should contain:
- name (str): Variable name
- type (str): Variable type (e.g., "String", "Number", "Boolean")
- value (Any): Variable value
- note (str, optional): Optional note/description for the variable
Example: [{"name": "var1", "type": "String", "value": "test", "note": "A test variable"}]
value_sets : Dict[str, Dict], optional
Dictionary of value sets where key is value set name and value contains:
- variableOverrides (List[Dict]): List of variable overrides
Example: {"ProductionSet": {"variableOverrides": [{"name": "var1", "value": "prod_value"}]}}
value_sets_order : List[str], optional
List of value set names in the order they should be applied.
If not provided, defaults to the keys of the value_sets dictionary.
workspace : str | uuid.UUID, optional
The Fabric workspace name or ID.
Defaults to None which resolves to the workspace of the attached lakehouse
or if no lakehouse attached, resolves to the workspace of the notebook.

Returns
-------
str
Status message indicating the result of the update operation.
"""

workspace_id = resolve_workspace_id(workspace)
variable_library_id = resolve_item_id(
item=variable_library, type="VariableLibrary", workspace=workspace
)

# Create the definition using the existing helper function
definition = _create_variable_library_definition(
variables=variables, value_sets=value_sets, value_sets_order=value_sets_order
)

payload = {"definition": definition}

# Build the request URL with optional query parameter
url = f"/v1/workspaces/{workspace_id}/VariableLibraries/{variable_library_id}/updateDefinition"

response = _base_api(
request=url,
method="post",
client="fabric_sp",
payload=payload,
status_codes=[200, 202],
lro_return_json=False,
)

if response.status_code == 200:
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

as before, don't need these if statements.

print(f"{icons.green_dot} Variable library definition updated successfully.")
return "Updated"
elif response.status_code == 202:
print(f"{icons.in_progress} Variable library definition update is in progress.")
return "In Progress"
else:
print(
f"{icons.red_dot} Failed to update variable library definition. Status code: {response.status_code}"
)
try:
result = response.json()
return f"Failed with status code {result.get('errorCode')} and message: {result.get('message')}"
except:
return f"Failed with status code {response.status_code}"