from datetime import datetime
from deprecated import deprecated
from pydantic import BaseModel, field_validator
from chariot import _apis, mcp_setting
from chariot.config import getLogger
from chariot_api._openapi.identity.api.organizations_api import OrganizationsApi
__all__ = [
"Tag",
"ParentProject",
"Project",
"ProjectDoesNotExistError",
"MultipleProjectsError",
"OrganizationDoesNotExistError",
"MultipleOrganizationsError",
"get_projects",
"create_project",
"get_project_id",
"get_project_users",
"organizations_enabled",
"get_organizations",
"get_organization_id",
"get_organization_name",
"PUBLIC_PROJECT",
"PRIVATE_PROJECT",
"RESTRICTED_PROJECT",
"INTERNAL_PROJECT",
]
logger = getLogger(__name__)
PUBLIC_PROJECT = "public"
PRIVATE_PROJECT = "private"
RESTRICTED_PROJECT = "restricted"
INTERNAL_PROJECT = "internal"
[docs]
class Tag(BaseModel):
key: str
value: str
[docs]
class ParentProject(BaseModel):
id: str
name: str
parent_id: str | None = None
[docs]
class Project(BaseModel):
id: str
name: str
description: str | None = None
created_at: datetime
updated_at: datetime
# New in organization mode
organization_id: str | None = None
visibility: str | None = None
# Deprecated in organization mode
parents: list[ParentProject] | None = None
tags: list[Tag] | None = None
@field_validator("created_at", "updated_at", mode="before")
@classmethod
def _parse_timestamp(cls, v):
return datetime.fromtimestamp(v / 1000.0)
class ProjectUser(BaseModel):
user_id: str | None = None
role: str | None = None
email: str | None = None
[docs]
class OrganizationDoesNotExistError(Exception):
def __init__(self, organization_name=None):
msg = f"Organization {organization_name!r} does not exist."
super().__init__(msg)
self.organization_name = organization_name
[docs]
class ProjectDoesNotExistError(Exception):
def __init__(self, project_name):
msg = f"Project {project_name!r} does not exist."
super().__init__(msg)
self.project_name = project_name
[docs]
class MultipleOrganizationsError(Exception):
def __init__(self, org_id_name_list):
self.org_id_name_list = org_id_name_list
msg = f"Multiple organizations found: {org_id_name_list}."
super().__init__(msg)
# todo: remove after bug fixed in projects microservice
[docs]
class MultipleProjectsError(Exception):
def __init__(self, project_list):
self.project_list = project_list
msg = f"Multiple projects found: {project_list}."
super().__init__(msg)
def organizations_api() -> OrganizationsApi:
return _apis.identity.organizations_api # type: ignore
[docs]
@deprecated(
version="0.25.0",
reason="Chariot always has organizations enabled, this function will be removed",
)
@_apis.login_required
def organizations_enabled() -> bool:
"""Returns True if chariot is running with organizations enabled"""
return True
[docs]
@_apis.login_required
def get_organizations() -> list[dict]:
"""List of organizations as dicts with name, id and is_public fields."""
orgs = organizations_api().v1_organizations_get().data
if orgs is None:
return []
return [
{
"name": o.name,
"id": o.id,
"is_public": o.is_public,
}
for o in orgs
]
[docs]
@_apis.login_required
def get_organization_id(organization_name: str | None = None) -> str:
if not organizations_enabled():
raise RuntimeError("organizations are not enabled - get_organization_id is not supported")
orgs = organizations_api().v1_organizations_get(name=organization_name).data
if orgs is None or len(orgs) == 0:
raise OrganizationDoesNotExistError(organization_name)
if len(orgs) > 1:
raise MultipleOrganizationsError([(p.id, p.name) for p in orgs])
return orgs[0].id
[docs]
@_apis.login_required
def get_organization_name(organization_id: str) -> str:
"""Return the org name from an `organization_id`."""
if not organizations_enabled():
raise RuntimeError("organizations are not enabled - get_organization_id is not supported")
org = organizations_api().v1_organizations_organization_id_get(organization_id).data
if org is None:
raise OrganizationDoesNotExistError(organization_id)
return org.name or ""
[docs]
@_apis.login_required
def get_projects(
*,
limit: int = 10,
offset: int = 0,
organization_id: str | None = None,
**kwargs,
) -> list[Project]:
if not organizations_enabled() and organization_id is not None:
raise ValueError("organizations are not enabled - organization_id is not supported")
response = _apis.identity.projects_api.v1_projects_get(
limit=limit, offset=offset, org=organization_id, **kwargs
)
return [Project.model_validate(p) for p in response.to_dict()["data"]]
[docs]
@mcp_setting(mutating=True)
@_apis.login_required
def create_project(
name: str,
description: str,
parent_id: str | None = None,
organization_id: str | None = None,
visibility: str | None = None,
) -> None:
if organizations_enabled():
if parent_id is not None:
raise ValueError("organizations are enabled - parent_id is not supported")
if organization_id is None or visibility is None:
raise ValueError(
"organizations are enabled - organization_id and visibility are required"
)
else:
if organization_id is not None or visibility is not None:
raise ValueError(
"organizations are not enabled - organization_id and visibility are not supported"
)
body = {
"name": name,
"description": description,
}
if organization_id is not None:
body["organization_id"] = organization_id
if visibility is not None:
body["visibility"] = visibility
if parent_id is not None:
body["parent_id"] = parent_id
_apis.identity.projects_api.v1_projects_post(body)
[docs]
@_apis.login_required
def get_project_id(
project_name: str | None = None,
organization_id: str | None = None,
) -> str:
"""Project id found corresponding to the given project name and organization id."""
return _get_project_id_org_mode(
project_name=project_name,
organization_id=organization_id,
)
[docs]
@_apis.login_required
def get_project_users(project_id: str, no_email: bool = False) -> list[ProjectUser]:
"""Get the list of users for a project by ID.
If no_email is True, results will not have emails populated. The query will be slightly faster
as well.
"""
resp = _apis.identity.projects_api.v1_projects_project_id_users_get(
project_id=project_id, no_email=no_email
)
return [ProjectUser.model_validate(d) for d in resp.to_dict()["data"] or []]
def _get_project_id_org_mode(
project_name: str | None = None,
organization_id: str | None = None,
) -> str:
# organization_id is optional
if project_name is None:
raise ValueError("project_name is required")
projs = _apis.identity.projects_api.v1_projects_get(
project_name=project_name, org=organization_id
).data
if projs is None or len(projs) == 0:
raise ProjectDoesNotExistError(project_name)
if len(projs) > 1:
raise MultipleProjectsError(
[{"id": p.id, "name": p.name, "org_id": p.organization_id} for p in projs]
)
return projs[0].id
def _get_project_id_from_project_args(
*,
project_id: str | None = None,
project_name: str | None = None,
organization_id: str | None = None,
) -> str:
if project_id is None and project_name is None:
raise ValueError("Either project_id or project_name must be specified.")
if project_id is not None and project_name is not None:
raise ValueError("Either project_id or project_name must be specified, not both.")
if project_id is not None:
return project_id
return get_project_id(project_name=project_name, organization_id=organization_id)