from datetime import datetime
from deprecated import deprecated
from pydantic import BaseModel, field_validator
from chariot import _apis
from chariot.config import getLogger
GLOBAL_PROJECT_ID = None
ORGANIZATIONS_ENABLED = None
__all__ = [
"Tag",
"ParentProject",
"Project",
"ProjectDoesNotExistError",
"MultipleProjectsError",
"OrganizationDoesNotExistError",
"MultipleOrganizationsError",
"get_projects",
"get_global_project_id",
"create_project",
"get_project_id",
"organizations_enabled",
"get_organization_id",
"PUBLIC_PROJECT",
"PRIVATE_PROJECT",
"RESTRICTED_PROJECT",
]
logger = getLogger(__name__)
PUBLIC_PROJECT = "public"
PRIVATE_PROJECT = "private"
RESTRICTED_PROJECT = "restricted"
[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)
[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, subproject_name=None):
if subproject_name is None:
msg = f"Project {project_name!r} does not exist."
else:
msg = f"Project {project_name}/{subproject_name} does not exist."
super().__init__(msg)
self.project_name = project_name
self.subproject_name = subproject_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_id_name_list):
self.project_id_name_list = project_id_name_list
msg = f"Multiple projects found: {project_id_name_list}."
super().__init__(msg)
[docs]
@_apis.login_required
def organizations_enabled() -> bool:
"""Returns True if chariot is running with organizations enabled"""
global ORGANIZATIONS_ENABLED
if ORGANIZATIONS_ENABLED is not None:
return ORGANIZATIONS_ENABLED
try:
# Handle any exception from the server not supporting this route
# To be super careful with versions of SDK vs Chariot
response = _apis.identity.organizations_api.organizations_enabled()
if response and response.data and isinstance(response.data.enabled, bool):
ORGANIZATIONS_ENABLED = response.data.enabled
else:
ORGANIZATIONS_ENABLED = False
except:
return False # Don't cache on exception so we can try again
return ORGANIZATIONS_ENABLED
[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 = _apis.identity.organizations_api.v1_organizations_get(name=organization_name).data
if 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_projects(
*,
limit: int = 10,
offset: int = 0,
organization_id: str | None = None,
) -> 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
)
return [Project.model_validate(p) for p in response.to_dict()["data"]]
[docs]
@deprecated(
version="0.16.0",
reason="Once Chariot migrates to organizations, there will no longer be a Global project, use organization + project name instead",
)
@_apis.login_required
def get_global_project_id() -> str:
"""Deprecated - Returns the id of the global project, will error if organizations are enabled"""
if organizations_enabled():
raise RuntimeError("organizations are enabled - there is no global project")
global GLOBAL_PROJECT_ID
if GLOBAL_PROJECT_ID is None:
global_proj = _apis.identity.projects_api.v1_projects_get(name="Global").data
global_proj = [p for p in global_proj if p.parents is None]
if len(global_proj) != 1:
raise RuntimeError("Error finding global project id.")
GLOBAL_PROJECT_ID = global_proj[0].id
return GLOBAL_PROJECT_ID
[docs]
@_apis.login_required
def create_project(
name: str,
description: str,
parent_id: str | None = None,
organization_id: str | None = None,
visibility: str | 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,
subproject_name: str | None = None,
organization_id: str | None = None,
) -> str:
"""Gets the project id from (sub)project names. If project_name is None then
it assumes the project is the global project
An organization is required when they are enabled and subproject_name is ignored.
"""
if organizations_enabled():
return _get_project_id_org_mode(
project_name=project_name,
subproject_name=subproject_name,
organization_id=organization_id,
)
else:
return _get_project_id_not_org_mode(
project_name=project_name,
subproject_name=subproject_name,
organization_id=organization_id,
)
def _get_project_id_org_mode(
project_name: str | None = None,
subproject_name: str | None = None,
organization_id: str | None = None,
) -> str:
# organization_id is optional
if project_name is None or subproject_name is not None:
raise ValueError(
"organizations are enabled - project_name is required, and subproject_name is not supported"
)
projs = _apis.identity.projects_api.v1_projects_get(
project_name=project_name, org=organization_id
).data
if len(projs) == 0:
raise ProjectDoesNotExistError(project_name)
if len(projs) > 1:
raise MultipleProjectsError([(p.id, p.name) for p in projs])
return projs[0].id
def _get_project_id_not_org_mode(
project_name: str | None = None,
subproject_name: str | None = None,
organization_id: str | None = None,
) -> str:
if organization_id is not None:
raise ValueError("organizations are not enabled - organization_id is not supported")
if project_name is None:
if subproject_name is not None:
raise ValueError(
"If specifying `subproject_name` then `project_name` must also be specified."
)
return get_global_project_id()
if subproject_name is None:
# handle special case of the Global project
if project_name == "Global":
return get_global_project_id()
projs = _apis.identity.projects_api.v1_projects_get(project_name=project_name).data
projs = [p for p in projs if p.name == project_name and len(p.parents) == 1]
if len(projs) == 0:
raise ProjectDoesNotExistError(project_name)
if len(projs) > 1:
raise MultipleProjectsError([(p.id, p.name) for p in projs])
return projs[0].id
projs = _apis.identity.projects_api.v1_projects_get(project_name=subproject_name).data
projs = [
p
for p in projs
if p.name == subproject_name and any([parent.name == project_name for parent in p.parents])
]
if len(projs) == 0:
raise ProjectDoesNotExistError(project_name, subproject_name)
if len(projs) > 1:
raise MultipleProjectsError([(p.id, p.name) 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,
subproject_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, subproject_name=subproject_name, organization_id=organization_id
)