api_lib.py

import numpy as np
from sysmlv2_client import SysMLV2Client, SysMLV2Error, SysMLV2NotFoundError
from typing import Any, Dict

def create_sysml_project(client: SysMLV2Client, name:str, description:str="project description"):
    created_project = None
    example_project_id = None
    commit_id = None
    new_project_data = {
        "@type": "Project",
        "name": name,
        "description": description
    }
    try:
        created_project = client.create_project(new_project_data)
        # Store the ID for later use
        example_project_id = created_project.get('@id')
    except SysMLV2Error as e:
        print(f"Error creating project: {e}")
    return created_project, example_project_id, commit_id


def get_project_by_name(client: SysMLV2Client, name: str):
    try:
        projects = client.get_projects()
        if projects:
            for project in projects:
                proj_id = project.get('@id', 'N/A')
                proj_name = project.get('name', 'N/A')
                if proj_name == name:
                    return project, proj_id
            # If no project matched
            return None, None
        else:
            print("  (No projects found)")
            return None, None
    except SysMLV2Error as e:
        print(f"Error getting projects: {e}")
        return None, None

def delete_project_data(client: SysMLV2Client, proj_id: str, branch_id: str = None):
    # Get the latest commit
    commits = client.list_commits(proj_id)
    if not commits:
        return None, None

    latest = commits[0] if isinstance(commits, list) else commits
    commit_id = latest.get("@id") or latest.get("id")
    if not commit_id:
        return None, None

    # Fetch elements for that commit
    elements = client.list_elements(proj_id, commit_id)
    print(f"_delete_project_data: Found {len(elements)} elements total")

    if not elements:
        return None, None

    # Filter out root namespace (no owningNamespace)
    elements_to_delete = [
        elem for elem in elements
        if "@id" in elem and elem.get("owningNamespace") is not None
    ]

    print(f"_delete_project_data: Deleting {len(elements_to_delete)} elements (excluding root)")

    change_payload = [
        {
            "identity": {
                "@id": elem["@id"]
            }
        }
        for elem in elements_to_delete
    ]

    # print (change_payload)

    if not change_payload:
        print("_delete_project_data: Nothing to delete.")
        return None, None

    commit = {
        "@type": "Commit",
        "description": "delete all elements except root namespace",
        "change": change_payload
    }

    try:
        resp = client.create_commit(proj_id, commit, branch_id=branch_id)
        commit1_id = resp.get("@id")
        if not commit1_id:
            print("*** WARNING: Could not extract commit ID ('@id') from response! ***")
            return None, None
        return resp, commit1_id

    except SysMLV2Error as e:
        print(f"Error creating delete commit: {e}")
        return None, None

# def delete_project_data (client: SysMLV2Client, proj_id:str, branch_id:str = None):
#     # Get the latest commit
#     commits = client.list_commits(proj_id)
#     if not commits:
#         return
#     latest = commits[0] if isinstance(commits, list) else commits
#     commit_id = latest.get("@id") or latest.get("id")
#     if not commit_id:
#         return

#     # Fetch elements for that commit
#     elements = client.list_elements(proj_id, commit_id)
#     print(f"_delete_project_data: Found {len(elements)} elements to delete")

#     if not elements:
#         return

#     # include elements in the change attribute which have an identity but no payload    
#     #     {
#     #         "@type": "Commit",
#     #         "description": "Commit 1: Create initial elements",
#     #         "change": [
#     #     {
#     #     "identity": {
#     #         "@id": "4b0d767c-318f-4160-8a5f-2a76a3c5a65a"
#     #     }
#     #     }
#     change_payload = [
#         {
#             "identity": {
#                 "@id": elem["@id"]
#             }
#         }
#         for elem in elements
#         if "@id" in elem
#     ]
#     commit = {"@type": "Commit", "description": f"delete all elements", "change": change_payload}
#     #print (commit)

#     try:
#         resp = client.create_commit(proj_id, commit, branch_id=branch_id)
#         commit1_id = resp.get('@id')
#         if not commit1_id:
#             print("\n*** WARNING: Could not extract commit ID ('@id') from response! ***")
#             return None, None
#         else:
#             return resp, commit1_id
#     except SysMLV2Error as e:
#         print(f"Error creating commit 1: {e}")
#         return None, None

def commit_to_project(client: SysMLV2Client, proj_id:str, payload:str, commit_msg:str = "default commit message", user_commit_data=None, delete_project_data:bool = False, branch_id=None,  replace_model:bool = False):
    if user_commit_data is None:
        commit1_data = {
            "@type": "Commit",
            "description": commit_msg,
            "change": payload
        }
    else:
        commit1_data = user_commit_data

    if delete_project_data:
        resp, commit_del_id = delete_project_data (client, proj_id, branch_id=branch_id)
        if not commit_del_id:
            print("\n*** WARNING: Could not delete project data before commit ***")
            return None, None

    try:
        commit1_response = client.create_commit(proj_id, commit1_data, branch_id=branch_id, replace=replace_model)
        commit1_id = commit1_response.get('@id')
        if not commit1_id:
            print("\n*** WARNING: Could not extract commit ID ('@id') from response! ***")
            return None, None
        else:
            return commit1_response, commit1_id
    except SysMLV2Error as e:
        print(f"Error creating commit 1: {e}")
        return None, None


def _get_default_branch_id(project: Dict[str, Any]) -> str | None:
    default_branch = project.get('defaultBranch')
    if isinstance(default_branch, dict):
        return default_branch.get('@id')
    return None

def _get_referenced_commit_id(branch: Dict[str, Any]) -> str | None:
    referenced_commit = branch.get('referencedCommit')
    if isinstance(referenced_commit, dict):
        return referenced_commit.get('@id')
    return None


def get_last_commit_from_project(client: SysMLV2Client, proj_obj: Dict[str, Any]):
    default_branch_id = _get_default_branch_id(proj_obj)
    proj_id = proj_obj.get('@id')
    response_data = client._request(method="GET", endpoint=f"/projects/{proj_id}/branches/{default_branch_id}", expected_status=200)
    commit_id = _get_referenced_commit_id(response_data)
    return commit_id

def create_branch(client: SysMLV2Client, proj_id:str, commit_id:str, branch_name:str):
    #create branch from that commit
    branch_data = {
#        "@type": "Branch",
        "name": branch_name,
        "head": {"@id": commit_id}
    }
    print (branch_data)
    try:
        created_branch = client.create_branch(proj_id, branch_data)
        example_branch_id = created_branch.get('@id')
        if not example_branch_id:
            print("\n*** WARNING: Could not extract branch ID ('@id') from response! ***")
            return None, None
        else:
            return created_branch, example_branch_id
    except SysMLV2Error as e:
        print(f"Error creating branch: {e}")
        return None, None