Skip to content

Commit d322eaa

Browse files
GG-YanneGG-Yanne
committed
feat(client): add teams related endpoints
1 parent ddd3401 commit d322eaa

15 files changed

+1270
-3
lines changed

‎pygitguardian/client.py

Lines changed: 236 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,9 @@
2626
)
2727
from .models import (
2828
APITokensResponse,
29+
CreateTeam,
30+
CreateTeamInvitation,
31+
CreateTeamMember,
2932
CursorPaginatedResponse,
3033
DeleteMember,
3134
Detail,
@@ -45,7 +48,14 @@
4548
SecretIncident,
4649
SecretScanPreferences,
4750
ServerMetadata,
51+
Team,
52+
TeamInvitation,
53+
TeamInvitationParameter,
54+
TeamMember,
55+
TeamMemberParameter,
56+
TeamsParameter,
4857
UpdateMember,
58+
UpdateTeam,
4959
)
5060
from .sca_models import (
5161
ComputeSCAFilesResult,
@@ -120,6 +130,14 @@ def is_create_ok(resp: Response) -> bool:
120130
)
121131

122132

133+
def is_delete_ok(resp: Response) -> bool:
134+
"""
135+
is_delete_ok returns True if the API returns code 204
136+
and the content type is JSON.
137+
"""
138+
return resp.status_code == codes.no_content
139+
140+
123141
def _create_tar(root_path: Path, filenames: List[str]) -> bytes:
124142
"""
125143
:param root_path: the root_path from which the tar is created
@@ -975,7 +993,223 @@ def delete_member(
975993
)
976994

977995
# We bypass `is_ok` because the response content type is none
978-
if response.status_code == 204:
979-
return 204
996+
if is_delete_ok(response):
997+
return response.status_code
998+
999+
return load_detail(response)
1000+
1001+
def list_teams(
1002+
self,
1003+
parameters: Optional[TeamsParameter] = None,
1004+
extra_headers: Optional[Dict[str, str]] = None,
1005+
) -> Union[Detail, CursorPaginatedResponse[Team]]:
1006+
response = self.get(
1007+
endpoint="teams",
1008+
data=parameters.to_dict() if parameters else {},
1009+
extra_headers=extra_headers,
1010+
)
1011+
1012+
obj: Union[Detail, CursorPaginatedResponse[Team]]
1013+
if is_ok(response):
1014+
obj = CursorPaginatedResponse[Team].from_response(response, Team)
1015+
else:
1016+
obj = load_detail(response)
1017+
1018+
obj.status_code
1019+
return obj
1020+
1021+
def get_team(
1022+
self,
1023+
team_id: int,
1024+
extra_headers: Optional[Dict[str, str]] = None,
1025+
) -> Union[Detail, Team]:
1026+
response = self.get(
1027+
endpoint=f"teams/{team_id}",
1028+
extra_headers=extra_headers,
1029+
)
1030+
1031+
obj: Union[Detail, Team]
1032+
if is_ok(response):
1033+
obj = Team.from_dict(response.json())
1034+
else:
1035+
obj = load_detail(response)
1036+
1037+
obj.status_code = response.status_code
1038+
return obj
1039+
1040+
def create_team(
1041+
self,
1042+
team: CreateTeam,
1043+
extra_headers: Optional[Dict[str, str]] = None,
1044+
) -> Union[Detail, Team]:
1045+
response = self.post(
1046+
endpoint="teams", data=CreateTeam.to_dict(team), extra_headers=extra_headers
1047+
)
1048+
1049+
obj: Union[Detail, Team]
1050+
if is_create_ok(response):
1051+
obj = Team.from_dict(response.json())
1052+
else:
1053+
obj = load_detail(response)
1054+
1055+
obj.status_code = response.status_code
1056+
return obj
1057+
1058+
def update_team(
1059+
self,
1060+
team: UpdateTeam,
1061+
extra_headers: Optional[Dict[str, str]] = None,
1062+
) -> Union[Detail, Team]:
1063+
team_id = team.id
1064+
data = UpdateTeam.to_dict(team)
1065+
del data["id"]
1066+
1067+
response = self.patch(
1068+
endpoint=f"teams/{team_id}",
1069+
data=data,
1070+
extra_headers=extra_headers,
1071+
)
1072+
1073+
obj: Union[Detail, Team]
1074+
if is_ok(response):
1075+
obj = Team.from_dict(response.json())
1076+
else:
1077+
obj = load_detail(response)
1078+
1079+
obj.status_code = response.status_code
1080+
return obj
1081+
1082+
def delete_team(
1083+
self,
1084+
team_id: int,
1085+
extra_headers: Optional[Dict[str, str]] = None,
1086+
) -> Union[Detail, int]:
1087+
response = self.delete(
1088+
endpoint=f"teams/{team_id}",
1089+
extra_headers=extra_headers,
1090+
)
1091+
1092+
if is_delete_ok(response):
1093+
return response.status_code
1094+
1095+
return load_detail(response)
1096+
1097+
def list_team_invitations(
1098+
self,
1099+
team_id: int,
1100+
parameters: Optional[TeamInvitationParameter] = None,
1101+
extra_headers: Optional[Dict[str, str]] = None,
1102+
) -> Union[Detail, CursorPaginatedResponse[TeamInvitation]]:
1103+
response = self.get(
1104+
endpoint=f"teams/{team_id}/team_invitations",
1105+
data=parameters.to_dict() if parameters else {},
1106+
extra_headers=extra_headers,
1107+
)
1108+
1109+
obj: Union[Detail, CursorPaginatedResponse[TeamInvitation]]
1110+
if is_ok(response):
1111+
obj = CursorPaginatedResponse[TeamInvitation].from_response(
1112+
response, TeamInvitation
1113+
)
1114+
else:
1115+
obj = load_detail(response)
1116+
1117+
obj.status_code
1118+
return obj
1119+
1120+
def create_team_invitation(
1121+
self,
1122+
team_id: int,
1123+
invitation: CreateTeamInvitation,
1124+
extra_headers: Optional[Dict[str, str]] = None,
1125+
) -> Union[Detail, TeamInvitation]:
1126+
response = self.post(
1127+
endpoint=f"teams/{team_id}/team_invitations",
1128+
data=CreateTeamInvitation.to_dict(invitation),
1129+
extra_headers=extra_headers,
1130+
)
1131+
1132+
obj: Union[Detail, TeamInvitation]
1133+
if is_create_ok(response):
1134+
obj = TeamInvitation.from_dict(response.json())
1135+
else:
1136+
obj = load_detail(response)
1137+
1138+
obj.status_code = response.status_code
1139+
return obj
1140+
1141+
def delete_team_invitation(
1142+
self,
1143+
team_id: int,
1144+
invitation_id: int,
1145+
extra_headers: Optional[Dict[str, str]] = None,
1146+
) -> Union[Detail, int]:
1147+
response = self.delete(
1148+
endpoint=f"teams/{team_id}/team_invitations/{invitation_id}",
1149+
extra_headers=extra_headers,
1150+
)
1151+
1152+
if is_delete_ok(response):
1153+
return response.status_code
1154+
1155+
return load_detail(response)
1156+
1157+
def list_team_members(
1158+
self,
1159+
team_id: int,
1160+
parameters: Optional[TeamMemberParameter] = None,
1161+
extra_headers: Optional[Dict[str, str]] = None,
1162+
) -> Union[Detail, CursorPaginatedResponse[TeamMember]]:
1163+
response = self.get(
1164+
endpoint=f"teams/{team_id}/team_memberships",
1165+
data=parameters.to_dict() if parameters else {},
1166+
extra_headers=extra_headers,
1167+
)
1168+
1169+
obj: Union[Detail, CursorPaginatedResponse[TeamMember]]
1170+
if is_ok(response):
1171+
obj = CursorPaginatedResponse[TeamMember].from_response(
1172+
response, TeamMember
1173+
)
1174+
else:
1175+
obj = load_detail(response)
1176+
1177+
obj.status_code
1178+
return obj
1179+
1180+
def create_team_member(
1181+
self,
1182+
team_id: int,
1183+
member: CreateTeamMember,
1184+
extra_headers: Optional[Dict[str, str]] = None,
1185+
) -> Union[Detail, TeamMember]:
1186+
response = self.post(
1187+
endpoint=f"teams/{team_id}/team_memberships",
1188+
data=CreateTeamMember.to_dict(member),
1189+
extra_headers=extra_headers,
1190+
)
1191+
1192+
obj: Union[Detail, TeamMember]
1193+
if is_create_ok(response):
1194+
obj = TeamMember.from_dict(response.json())
1195+
else:
1196+
obj = load_detail(response)
1197+
1198+
obj.status_code = response.status_code
1199+
return obj
1200+
1201+
def delete_team_member(
1202+
self,
1203+
team_id: int,
1204+
team_member_id: int,
1205+
extra_headers: Optional[Dict[str, str]] = None,
1206+
) -> Union[Detail, int]:
1207+
response = self.delete(
1208+
endpoint=f"teams/{team_id}/team_memberships/{team_member_id}",
1209+
extra_headers=extra_headers,
1210+
)
1211+
1212+
if is_delete_ok(response):
1213+
return response.status_code
9801214

9811215
return load_detail(response)

‎pygitguardian/config.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
DEFAULT_BASE_URI = "https://api.gitguardian.com"
1+
DEFAULT_BASE_URI = "http://localhost:3000/exposed"
22
DEFAULT_API_VERSION = "v1"
33
DEFAULT_TIMEOUT = 20.0 # 20s default timeout
44

0 commit comments

Comments
 (0)