Coverage for src/hush_hush/_generated/api/tokens/create_consumer_token.py: 0%

1# Code generated by openapi-python-client from hush-hush/api/openapi.yaml; DO NOT EDIT. 

2 

3from http import HTTPStatus 

4from typing import Any 

5 

6import httpx 

7 

8from ... import errors 

9from ...client import AuthenticatedClient, Client 

10from ...models.consumer_token_with_value import ConsumerTokenWithValue 

11from ...models.create_consumer_token_request import CreateConsumerTokenRequest 

12from ...models.error import Error 

13from ...types import UNSET, Response, Unset 

14 

15 

16def _get_kwargs( 

17 *, 

18 body: CreateConsumerTokenRequest, 

19 x_csrf_token: str | Unset = UNSET, 

20) -> dict[str, Any]: 

21 headers: dict[str, Any] = {} 

22 if not isinstance(x_csrf_token, Unset): 

23 headers["X-CSRF-Token"] = x_csrf_token 

24 

25 _kwargs: dict[str, Any] = { 

26 "method": "post", 

27 "url": "/consumer-tokens", 

28 } 

29 

30 _kwargs["json"] = body.to_dict() 

31 

32 headers["Content-Type"] = "application/json" 

33 

34 _kwargs["headers"] = headers 

35 return _kwargs 

36 

37 

38def _parse_response( 

39 *, client: AuthenticatedClient | Client, response: httpx.Response 

40) -> ConsumerTokenWithValue | Error | None: 

41 if response.status_code == 201: 

42 response_201 = ConsumerTokenWithValue.from_dict(response.json()) 

43 

44 return response_201 

45 

46 if response.status_code == 400: 

47 response_400 = Error.from_dict(response.json()) 

48 

49 return response_400 

50 

51 if response.status_code == 401: 

52 response_401 = Error.from_dict(response.json()) 

53 

54 return response_401 

55 

56 if response.status_code == 413: 

57 response_413 = Error.from_dict(response.json()) 

58 

59 return response_413 

60 

61 if response.status_code == 422: 

62 response_422 = Error.from_dict(response.json()) 

63 

64 return response_422 

65 

66 if client.raise_on_unexpected_status: 

67 raise errors.UnexpectedStatus(response.status_code, response.content) 

68 else: 

69 return None 

70 

71 

72def _build_response( 

73 *, client: AuthenticatedClient | Client, response: httpx.Response 

74) -> Response[ConsumerTokenWithValue | Error]: 

75 return Response( 

76 status_code=HTTPStatus(response.status_code), 

77 content=response.content, 

78 headers=response.headers, 

79 parsed=_parse_response(client=client, response=response), 

80 ) 

81 

82 

83def sync_detailed( 

84 *, 

85 client: AuthenticatedClient, 

86 body: CreateConsumerTokenRequest, 

87 x_csrf_token: str | Unset = UNSET, 

88) -> Response[ConsumerTokenWithValue | Error]: 

89 """Create a consumer read token 

90 

91 Issues a new read token scoped to one consumer name. The raw 

92 token value is returned here only - it is never recoverable 

93 again once this response is read. A write bearer token may mint 

94 one too, not only an admin session - a narrowing delegation, 

95 since the resulting token can only ever read as far as its one 

96 bound consumer, never as far as the write token that issued it 

97 (alrayyes/hush-hush#467). A session-authenticated call needs its 

98 CSRF token too; a bearer-token-authenticated one doesn't, since 

99 there's no session to have one. 

100 

101 Args: 

102 x_csrf_token (str | Unset): 

103 body (CreateConsumerTokenRequest): 

104 

105 Raises: 

106 errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. 

107 httpx.TimeoutException: If the request takes longer than Client.timeout. 

108 

109 Returns: 

110 Response[ConsumerTokenWithValue | Error] 

111 """ 

112 

113 kwargs = _get_kwargs( 

114 body=body, 

115 x_csrf_token=x_csrf_token, 

116 ) 

117 

118 response = client.get_httpx_client().request( 

119 **kwargs, 

120 ) 

121 

122 return _build_response(client=client, response=response) 

123 

124 

125def sync( 

126 *, 

127 client: AuthenticatedClient, 

128 body: CreateConsumerTokenRequest, 

129 x_csrf_token: str | Unset = UNSET, 

130) -> ConsumerTokenWithValue | Error | None: 

131 """Create a consumer read token 

132 

133 Issues a new read token scoped to one consumer name. The raw 

134 token value is returned here only - it is never recoverable 

135 again once this response is read. A write bearer token may mint 

136 one too, not only an admin session - a narrowing delegation, 

137 since the resulting token can only ever read as far as its one 

138 bound consumer, never as far as the write token that issued it 

139 (alrayyes/hush-hush#467). A session-authenticated call needs its 

140 CSRF token too; a bearer-token-authenticated one doesn't, since 

141 there's no session to have one. 

142 

143 Args: 

144 x_csrf_token (str | Unset): 

145 body (CreateConsumerTokenRequest): 

146 

147 Raises: 

148 errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. 

149 httpx.TimeoutException: If the request takes longer than Client.timeout. 

150 

151 Returns: 

152 ConsumerTokenWithValue | Error 

153 """ 

154 

155 return sync_detailed( 

156 client=client, 

157 body=body, 

158 x_csrf_token=x_csrf_token, 

159 ).parsed 

160 

161 

162async def asyncio_detailed( 

163 *, 

164 client: AuthenticatedClient, 

165 body: CreateConsumerTokenRequest, 

166 x_csrf_token: str | Unset = UNSET, 

167) -> Response[ConsumerTokenWithValue | Error]: 

168 """Create a consumer read token 

169 

170 Issues a new read token scoped to one consumer name. The raw 

171 token value is returned here only - it is never recoverable 

172 again once this response is read. A write bearer token may mint 

173 one too, not only an admin session - a narrowing delegation, 

174 since the resulting token can only ever read as far as its one 

175 bound consumer, never as far as the write token that issued it 

176 (alrayyes/hush-hush#467). A session-authenticated call needs its 

177 CSRF token too; a bearer-token-authenticated one doesn't, since 

178 there's no session to have one. 

179 

180 Args: 

181 x_csrf_token (str | Unset): 

182 body (CreateConsumerTokenRequest): 

183 

184 Raises: 

185 errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. 

186 httpx.TimeoutException: If the request takes longer than Client.timeout. 

187 

188 Returns: 

189 Response[ConsumerTokenWithValue | Error] 

190 """ 

191 

192 kwargs = _get_kwargs( 

193 body=body, 

194 x_csrf_token=x_csrf_token, 

195 ) 

196 

197 response = await client.get_async_httpx_client().request(**kwargs) 

198 

199 return _build_response(client=client, response=response) 

200 

201 

202async def asyncio( 

203 *, 

204 client: AuthenticatedClient, 

205 body: CreateConsumerTokenRequest, 

206 x_csrf_token: str | Unset = UNSET, 

207) -> ConsumerTokenWithValue | Error | None: 

208 """Create a consumer read token 

209 

210 Issues a new read token scoped to one consumer name. The raw 

211 token value is returned here only - it is never recoverable 

212 again once this response is read. A write bearer token may mint 

213 one too, not only an admin session - a narrowing delegation, 

214 since the resulting token can only ever read as far as its one 

215 bound consumer, never as far as the write token that issued it 

216 (alrayyes/hush-hush#467). A session-authenticated call needs its 

217 CSRF token too; a bearer-token-authenticated one doesn't, since 

218 there's no session to have one. 

219 

220 Args: 

221 x_csrf_token (str | Unset): 

222 body (CreateConsumerTokenRequest): 

223 

224 Raises: 

225 errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. 

226 httpx.TimeoutException: If the request takes longer than Client.timeout. 

227 

228 Returns: 

229 ConsumerTokenWithValue | Error 

230 """ 

231 

232 return ( 

233 await asyncio_detailed( 

234 client=client, 

235 body=body, 

236 x_csrf_token=x_csrf_token, 

237 ) 

238 ).parsed