Forum Discussion
Sidhant
1 year agoAdvocate V
Power BI Report creation through Script
Hi everyone, So recently I was experimenting that can we create reports through script apart from Power BI desktop and I found out that we can do it by using the '.pbip' file extension. And I did fo...
- 1 year ago
Hi everyone,
A quick update so I am able to create reports in a programatic fashion it looks like:def create_model_bim(tables, output_folder): # Load a fresh copy of the model template model_bim = copy.deepcopy(model_template) # Replace {{query_order}} annotation for annotation in model_bim["model"].get("annotations", []): if annotation.get("value") == "{{query_order}}": annotation["value"] = str([tbl["name"] for tbl in tables]) # Build table_column_map to validate relationships later table_column_map = {} with open("Static/data_type_mapping.json") as f: type_mapping = json.load(f) model_bim["model"]["tables"] = [] for table in tables: column_names = [col["name"] for col in table["columns"]] table_dict = { "name": table["name"], "columns": [ { "name": col["name"], "dataType": type_mapping.get(col["type"].lower(), "string"), "sourceColumn": col["name"] } for col in table["columns"] ], "partitions": [ { "name": f"{table['name']}_Partition", "mode": "import", "source": { "type": "m", "expression": ( f"let Source = Sql.Database(\"server_name\", \"database_name\") " f"in Source{{[Schema=\"dbo\", Item=\"{table['name']}\"]}}[Data]" ) } } ] } model_bim["model"]["tables"].append(table_dict) # table_column_map[table["name"]] = [col["name"] for col in table["columns"]] table_column_map[table["name"]] = column_names # ✅ Load relationships sheet and parse valid entries # metadata_path = os.path.join(os.path.dirname(output_folder), "Files", "MetaData.xlsx") metadata_path = os.path.abspath(os.path.join(output_folder, "..", "..", "Files", "MetaData.xlsx")) # relationships = extract_relationships_from_metadata(metadata_path) relationships = extract_relationships_from_metadata(metadata_path, table_column_map) if relationships: model_bim["model"]["relationships"] = relationships measures = extract_measures_from_metadata(metadata_path) if measures: measures_table = { "name": "__Measures", "columns": [ { "name": "Dummy", "dataType": "string" } ], "measures": measures, "partitions": [ { "name": "__Measures_Partition", "mode": "import", "source": { "type": "m", "expression": "let Source = #table({\"Dummy\"}, {}) in Source" } } ] } model_bim["model"]["tables"].append(measures_table) # Write model.bim to file bim_path = os.path.join(output_folder, "model.bim") with open(bim_path, "w", encoding="utf-8") as file: json.dump(model_bim, file, indent=4) print(f"✅ model.bim created at: {bim_path}") #-- Report.json file creation: def create_valid_report_json(report_folder_path, chart_types_df, chart_axes_df): base_config = copy.deepcopy(report_static_config["config"]) def create_visual_config(visual_type, visual_index, axes): visual_id = str(uuid.uuid4()) layout = { "id": 0, "position": { "x": 100.0 + (visual_index % 2) * 450.0, "y": 100.0 + (visual_index // 2) * 350.0, "z": 0, "width": 400.0, "height": 300.0, "tabOrder": 0 } } projections = {} selects = [] from_tables = set() for axis_type, axis_list in axes.items(): projections[axis_type] = [] for axis in axis_list: table = axis['table_name'] column = axis['column'] full_ref = f"{table}.{column}" from_tables.add(table) if axis.get('aggregation'): agg_func = 0 # sum projections[axis_type].append({"queryRef": f"Sum({full_ref})"}) selects.append({ "Aggregation": { "Expression": { "Column": { "Expression": {"SourceRef": {"Source": table}}, "Property": column } }, "Function": agg_func }, "Name": f"Sum({full_ref})", "NativeReferenceName": f"Sum of {column}" }) else: projections[axis_type].append({"queryRef": full_ref, "active": True}) selects.append({ "Column": { "Expression": {"SourceRef": {"Source": table}}, "Property": column }, "Name": full_ref, "NativeReferenceName": column }) visual_config = { "name": visual_id, "layouts": [layout], "singleVisual": { "visualType": visual_type, "projections": projections, "prototypeQuery": { "Version": 2, "From": [{"Name": t, "Entity": t, "Type": 0} for t in from_tables], "Select": selects }, "drillFilterOtherVisuals": True, "hasDefaultSort": True, "objects": {}, "vcObjects": { "title": [ { "properties": { "text": { "expr": { "Literal": { "Value": f"'{visual_type.title()} Visual {visual_index + 1}'" } } } } } ] } } } return visual_config visual_containers = [] for idx, row in chart_types_df.iterrows(): visual_type = row.get('plot_type') worksheet = row.get('worksheet') if not visual_type or not worksheet: continue # Skip incomplete rows relevant_axes = chart_axes_df[ (chart_axes_df['worksheet'] == worksheet) & (chart_axes_df['type'].isin(['rows', 'cols'])) & # (chart_axes_df['order_id'] == 0) & (chart_axes_df['table_name'].notna()) ] axes_dict = {} for axis_type in ['rows', 'cols']: group = relevant_axes[relevant_axes['type'] == axis_type] if not group.empty: axes_dict['Category' if axis_type == 'rows' else 'Y'] = group.apply( lambda axis_row: { 'table_name': axis_row['table_name'], 'column': axis_row['column'], 'aggregation': str(axis_row.get('aggregation', '')).strip().lower() == 'sum' }, axis=1 ).tolist() if axes_dict: config = create_visual_config(visual_type, idx, axes_dict) container = { "config": json.dumps(config), "filters": "[]", "height": 300.0, "width": 400.0, "x": 100.0 + (idx % 2) * 450.0, "y": 100.0 + (idx // 2) * 350.0, "z": 0.0 } visual_containers.append(container) report_json = { "config": json.dumps(base_config), "layoutOptimization": 0, "resourcePackages": [], "sections": [ { "config": "{}", "displayName": "Auto Page", "displayOption": 1, "filters": "[]", "height": 720.0, "name": str(uuid.uuid4()), "visualContainers": visual_containers, "width": 1280.0 } ] } output_path = os.path.join(report_folder_path, "report.json") with open(output_path, "w", encoding="utf-8") as f: json.dump(report_json, f, indent=4) print(f"✅ report.json with {len(visual_containers)} visuals written at: {output_path}")
Sidhant
1 year agoAdvocate V
Hi everyone,
A quick update so I am able to create reports in a programatic fashion it looks like:
def create_model_bim(tables, output_folder):
# Load a fresh copy of the model template
model_bim = copy.deepcopy(model_template)
# Replace {{query_order}} annotation
for annotation in model_bim["model"].get("annotations", []):
if annotation.get("value") == "{{query_order}}":
annotation["value"] = str([tbl["name"] for tbl in tables])
# Build table_column_map to validate relationships later
table_column_map = {}
with open("Static/data_type_mapping.json") as f:
type_mapping = json.load(f)
model_bim["model"]["tables"] = []
for table in tables:
column_names = [col["name"] for col in table["columns"]]
table_dict = {
"name": table["name"],
"columns": [
{
"name": col["name"],
"dataType": type_mapping.get(col["type"].lower(), "string"),
"sourceColumn": col["name"]
} for col in table["columns"]
],
"partitions": [
{
"name": f"{table['name']}_Partition",
"mode": "import",
"source": {
"type": "m",
"expression": (
f"let Source = Sql.Database(\"server_name\", \"database_name\") "
f"in Source{{[Schema=\"dbo\", Item=\"{table['name']}\"]}}[Data]"
)
}
}
]
}
model_bim["model"]["tables"].append(table_dict)
# table_column_map[table["name"]] = [col["name"] for col in table["columns"]]
table_column_map[table["name"]] = column_names
# ✅ Load relationships sheet and parse valid entries
# metadata_path = os.path.join(os.path.dirname(output_folder), "Files", "MetaData.xlsx")
metadata_path = os.path.abspath(os.path.join(output_folder, "..", "..", "Files", "MetaData.xlsx"))
# relationships = extract_relationships_from_metadata(metadata_path)
relationships = extract_relationships_from_metadata(metadata_path, table_column_map)
if relationships:
model_bim["model"]["relationships"] = relationships
measures = extract_measures_from_metadata(metadata_path)
if measures:
measures_table = {
"name": "__Measures",
"columns": [
{
"name": "Dummy",
"dataType": "string"
}
],
"measures": measures,
"partitions": [
{
"name": "__Measures_Partition",
"mode": "import",
"source":
{
"type": "m",
"expression": "let Source = #table({\"Dummy\"}, {}) in Source"
}
}
]
}
model_bim["model"]["tables"].append(measures_table)
# Write model.bim to file
bim_path = os.path.join(output_folder, "model.bim")
with open(bim_path, "w", encoding="utf-8") as file:
json.dump(model_bim, file, indent=4)
print(f"✅ model.bim created at: {bim_path}")
#-- Report.json file creation:
def create_valid_report_json(report_folder_path, chart_types_df, chart_axes_df):
base_config = copy.deepcopy(report_static_config["config"])
def create_visual_config(visual_type, visual_index, axes):
visual_id = str(uuid.uuid4())
layout = {
"id": 0,
"position": {
"x": 100.0 + (visual_index % 2) * 450.0,
"y": 100.0 + (visual_index // 2) * 350.0,
"z": 0,
"width": 400.0,
"height": 300.0,
"tabOrder": 0
}
}
projections = {}
selects = []
from_tables = set()
for axis_type, axis_list in axes.items():
projections[axis_type] = []
for axis in axis_list:
table = axis['table_name']
column = axis['column']
full_ref = f"{table}.{column}"
from_tables.add(table)
if axis.get('aggregation'):
agg_func = 0 # sum
projections[axis_type].append({"queryRef": f"Sum({full_ref})"})
selects.append({
"Aggregation": {
"Expression": {
"Column": {
"Expression": {"SourceRef": {"Source": table}},
"Property": column
}
},
"Function": agg_func
},
"Name": f"Sum({full_ref})",
"NativeReferenceName": f"Sum of {column}"
})
else:
projections[axis_type].append({"queryRef": full_ref, "active": True})
selects.append({
"Column": {
"Expression": {"SourceRef": {"Source": table}},
"Property": column
},
"Name": full_ref,
"NativeReferenceName": column
})
visual_config = {
"name": visual_id,
"layouts": [layout],
"singleVisual": {
"visualType": visual_type,
"projections": projections,
"prototypeQuery": {
"Version": 2,
"From": [{"Name": t, "Entity": t, "Type": 0} for t in from_tables],
"Select": selects
},
"drillFilterOtherVisuals": True,
"hasDefaultSort": True,
"objects": {},
"vcObjects": {
"title": [
{
"properties": {
"text": {
"expr": {
"Literal": {
"Value": f"'{visual_type.title()} Visual {visual_index + 1}'"
}
}
}
}
}
]
}
}
}
return visual_config
visual_containers = []
for idx, row in chart_types_df.iterrows():
visual_type = row.get('plot_type')
worksheet = row.get('worksheet')
if not visual_type or not worksheet:
continue # Skip incomplete rows
relevant_axes = chart_axes_df[
(chart_axes_df['worksheet'] == worksheet) &
(chart_axes_df['type'].isin(['rows', 'cols'])) &
# (chart_axes_df['order_id'] == 0) &
(chart_axes_df['table_name'].notna())
]
axes_dict = {}
for axis_type in ['rows', 'cols']:
group = relevant_axes[relevant_axes['type'] == axis_type]
if not group.empty:
axes_dict['Category' if axis_type == 'rows' else 'Y'] = group.apply(
lambda axis_row: {
'table_name': axis_row['table_name'],
'column': axis_row['column'],
'aggregation': str(axis_row.get('aggregation', '')).strip().lower() == 'sum'
},
axis=1
).tolist()
if axes_dict:
config = create_visual_config(visual_type, idx, axes_dict)
container = {
"config": json.dumps(config),
"filters": "[]",
"height": 300.0,
"width": 400.0,
"x": 100.0 + (idx % 2) * 450.0,
"y": 100.0 + (idx // 2) * 350.0,
"z": 0.0
}
visual_containers.append(container)
report_json = {
"config": json.dumps(base_config),
"layoutOptimization": 0,
"resourcePackages": [],
"sections": [
{
"config": "{}",
"displayName": "Auto Page",
"displayOption": 1,
"filters": "[]",
"height": 720.0,
"name": str(uuid.uuid4()),
"visualContainers": visual_containers,
"width": 1280.0
}
]
}
output_path = os.path.join(report_folder_path, "report.json")
with open(output_path, "w", encoding="utf-8") as f:
json.dump(report_json, f, indent=4)
print(f"✅ report.json with {len(visual_containers)} visuals written at: {output_path}")