Skip to content
Merged
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
119 changes: 119 additions & 0 deletions examples/organization_tags.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,119 @@
#!/usr/bin/env python3
# Copyright IBM Corp. 2025, 2026
# SPDX-License-Identifier: MPL-2.0

"""Organization tags operations example.

Demonstrates:
1. list() - list tags in an organization
2. add_workspaces() - associate a workspace with a tag
3. delete() - delete a tag from an organization

Usage:
python examples/organization_tags.py --org my-org
python examples/organization_tags.py --org my-org --tag-id tag-abc123 --workspace-id ws-xyz
"""

from __future__ import annotations

import argparse
import os

from pytfe import TFEClient, TFEConfig
from pytfe.errors import TFEError
from pytfe.models.organization_tags import (
AddWorkspacesToTagOptions,
OrganizationTagsDeleteOptions,
)


def main() -> None:
parser = argparse.ArgumentParser(
description="Organization Tags demo for python-tfe SDK"
)
parser.add_argument(
"--address", default=os.getenv("TFE_ADDRESS", "https://app.terraform.io")
)
parser.add_argument("--token", default=os.getenv("TFE_TOKEN", ""))
parser.add_argument(
"--org",
default=os.getenv("TFE_ORG", ""),
help="Organization name",
)
parser.add_argument(
"--tag-id",
default=os.getenv("TFE_TAG_ID", ""),
help="Tag ID for add/delete operations",
)
parser.add_argument(
"--workspace-id",
default=os.getenv("TFE_WORKSPACE_ID", ""),
help="Workspace ID to associate with tag",
)
args = parser.parse_args()

if not args.token:
print("Error: TFE_TOKEN environment variable or --token required")
return

if not args.org:
print("Error: TFE_ORG environment variable or --org required")
return

cfg = TFEConfig(address=args.address, token=args.token)
client = TFEClient(cfg)

# 1) List tags
try:
print("[LIST] Listing organization tags")
print(f"[LIST] organization={args.org}")
tags = list(client.organization_tags.list(args.org))
print(f"[LIST] total_tags={len(tags)}")
for tag in tags:
print(
f"[LIST] id={tag.id}, name={tag.name}, instance_count={tag.instance_count}"
)
if not tags:
print("[LIST] no tags found")
except TFEError as exc:
print(f"[LIST] API error: {exc}")
return

if not args.tag_id:
print("[ADD_WORKSPACES] skipped: set --tag-id or TFE_TAG_ID")
print("[DELETE] skipped: set --tag-id or TFE_TAG_ID")
return

# 2) Add workspace to tag
if args.workspace_id:
print("[ADD_WORKSPACES] Associating a workspace to a tag")
print(
f"[ADD_WORKSPACES] organization={args.org}, tag_id={args.tag_id}, workspace_id={args.workspace_id}"
)
try:
client.organization_tags.add_workspaces(
args.org,
args.tag_id,
AddWorkspacesToTagOptions(workspace_ids=[args.workspace_id]),
)
print("[ADD_WORKSPACES] workspace associated")
except TFEError as exc:
print(f"[ADD_WORKSPACES] API error: {exc}")
else:
print("[ADD_WORKSPACES] skipped: set --workspace-id or TFE_WORKSPACE_ID")

# 3) Delete tag
print("[DELETE] Deleting a tag from the organization")
print(f"[DELETE] organization={args.org}, tag_id={args.tag_id}")
try:
client.organization_tags.delete(
args.org,
OrganizationTagsDeleteOptions(ids=[args.tag_id]),
)
print("[DELETE] tag deleted")
except TFEError as exc:
print(f"[DELETE] API error: {exc}")


if __name__ == "__main__":
main()
2 changes: 2 additions & 0 deletions src/pytfe/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
from .resources.oauth_client import OAuthClients
from .resources.oauth_token import OAuthTokens
from .resources.organization_membership import OrganizationMemberships
from .resources.organization_tags import OrganizationTags
from .resources.organization_token import OrganizationTokens
from .resources.organizations import Organizations
from .resources.plan import Plans
Expand Down Expand Up @@ -76,6 +77,7 @@ def __init__(self, config: TFEConfig | None = None):
self.organizations = Organizations(self._transport)
self.organization_memberships = OrganizationMemberships(self._transport)
self.users = Users(self._transport)
Comment thread
isivaselvan marked this conversation as resolved.
self.organization_tags = OrganizationTags(self._transport)
self.organization_tokens = OrganizationTokens(self._transport)
self.projects = Projects(self._transport)
self.variables = Variables(self._transport)
Expand Down
5 changes: 5 additions & 0 deletions src/pytfe/errors.py
Original file line number Diff line number Diff line change
Expand Up @@ -121,6 +121,11 @@ class ErrStateVersionUploadNotSupported(TFEError): ...
ERR_REQUIRED_TAG_KEY = "tag key is required"
ERR_INVALID_TAG_KEY = "invalid tag key"

# Organization Tag Error Constants
ERR_INVALID_TAG = "invalid value for tag"
ERR_REQUIRED_TAG_ID = "tag ID is required"
ERR_REQUIRED_TAG_WORKSPACE_ID = "workspace ID is required"


class WorkspaceNotFound(NotFound): ...

Expand Down
55 changes: 55 additions & 0 deletions src/pytfe/models/organization_tags.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,55 @@
# Copyright IBM Corp. 2025, 2026
# SPDX-License-Identifier: MPL-2.0

from __future__ import annotations

from pydantic import BaseModel, ConfigDict, Field

from .organization import Organization


class OrganizationTag(BaseModel):
"""Terraform Enterprise organization tag."""

model_config = ConfigDict(populate_by_name=True, extra="forbid")

id: str = Field(..., description="Tag ID")
name: str | None = Field(None, description="Tag name")
instance_count: int | None = Field(
None,
alias="instance-count",
description="Number of workspaces that have this tag",
)
organization: Organization | None = Field(
None,
description="Organization this tag belongs to",
)


class OrganizationTagsListOptions(BaseModel):
"""Options for listing organization tags."""

model_config = ConfigDict(populate_by_name=True, extra="forbid")

filter: str | None = Field(None, alias="filter[exclude][taggable][id]")
query: str | None = Field(
None,
alias="q",
description="Search query string for tag name likeness",
)


class OrganizationTagsDeleteOptions(BaseModel):
"""Options for deleting tags from an organization."""

model_config = ConfigDict(extra="forbid")

ids: list[str] = Field(default_factory=list)


class AddWorkspacesToTagOptions(BaseModel):
"""Options for associating workspaces with a tag."""

model_config = ConfigDict(extra="forbid")

workspace_ids: list[str] = Field(default_factory=list)
111 changes: 111 additions & 0 deletions src/pytfe/resources/organization_tags.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,111 @@
# Copyright IBM Corp. 2025, 2026
# SPDX-License-Identifier: MPL-2.0

from __future__ import annotations

from collections.abc import Iterator
from typing import Any
from urllib.parse import quote

from ..errors import (
ERR_INVALID_ORG,
ERR_INVALID_TAG,
ERR_REQUIRED_TAG_ID,
ERR_REQUIRED_TAG_WORKSPACE_ID,
)
from ..models.organization import Organization
from ..models.organization_tags import (
AddWorkspacesToTagOptions,
OrganizationTag,
OrganizationTagsDeleteOptions,
OrganizationTagsListOptions,
)
from ..utils import valid_string_id
from ._base import _Service


class OrganizationTags(_Service):
"""Organization tags service for Terraform Enterprise."""

def list(
self,
organization: str,
options: OrganizationTagsListOptions | None = None,
) -> Iterator[OrganizationTag]:
"""List all tags within an organization."""
if not valid_string_id(organization):
raise ValueError(ERR_INVALID_ORG)
return self._iter_tags(organization, options)

def _iter_tags(
self,
organization: str,
options: OrganizationTagsListOptions | None = None,
) -> Iterator[OrganizationTag]:
path = f"/api/v2/organizations/{quote(organization)}/tags"
params = options.model_dump(by_alias=True, exclude_none=True) if options else {}
for item in self._list(path, params=params):
yield self._parse_organization_tag(item)

def delete(
self,
organization: str,
options: OrganizationTagsDeleteOptions,
) -> None:
"""Delete tags from an organization."""
if not valid_string_id(organization):
raise ValueError(ERR_INVALID_ORG)

if len(options.ids) == 0:
raise ValueError(ERR_REQUIRED_TAG_ID)

for tag_id in options.ids:
if not valid_string_id(tag_id):
raise ValueError(f"{tag_id} is not a valid id value")

body = {"data": [{"type": "tags", "id": tag_id} for tag_id in options.ids]}
path = f"/api/v2/organizations/{quote(organization)}/tags"
self.t.request("DELETE", path, json_body=body)

def add_workspaces(
self, organization: str, tag: str, options: AddWorkspacesToTagOptions
) -> None:
"""Associate workspaces with an organization tag."""
if not valid_string_id(organization):
raise ValueError(ERR_INVALID_ORG)
if not valid_string_id(tag):
raise ValueError(ERR_INVALID_TAG)

if len(options.workspace_ids) == 0:
raise ValueError(ERR_REQUIRED_TAG_WORKSPACE_ID)

for workspace_id in options.workspace_ids:
if not valid_string_id(workspace_id):
raise ValueError(f"{workspace_id} is not a valid id value")

body = {
"data": [
{"type": "workspaces", "id": workspace_id}
for workspace_id in options.workspace_ids
]
}
path = f"/api/v2/tags/{quote(tag)}/relationships/workspaces"
self.t.request("POST", path, json_body=body)

def _parse_organization_tag(self, data: dict[str, Any]) -> OrganizationTag:
attributes = data.get("attributes", {})
relationships = data.get("relationships", {})

org = None
org_data = relationships.get("organization", {}).get("data")
if org_data and isinstance(org_data, dict):
org = Organization(id=org_data.get("id"))

return OrganizationTag.model_validate(
{
"id": data.get("id", ""),
"name": attributes.get("name"),
"instance-count": attributes.get("instance-count"),
"organization": org,
}
)
Loading
Loading