0
# API Client
1

2
HTTP client for HubSpot API communication providing authentication, error handling, retry logic, and custom object metadata discovery. Handles both OAuth 2.0 and Private App authentication methods.
3

4
## Capabilities
5

6
### Client Initialization
7

8
Creates an authenticated HTTP client for HubSpot API communication.
9

10
```python { .api }
11
class API:
12
    BASE_URL = "https://api.hubapi.com"
13
    USER_AGENT = "Airbyte"
14
    
15
    def __init__(self, credentials: Mapping[str, Any]):
16
        """
17
        Initialize API client with credentials.
18
        
19
        Parameters:
20
        - credentials: Authentication credentials (OAuth or Private App)
21
        """
22
```
23

24
### Authentication Type Detection
25

26
Determines the authentication method based on credentials.
27

28
```python { .api }
29
def is_oauth2(self) -> bool:
30
    """
31
    Check if using OAuth 2.0 authentication.
32
    
33
    Returns:
34
    - True if credentials are OAuth 2.0 type
35
    """
36

37
def is_private_app(self) -> bool:
38
    """
39
    Check if using Private App authentication.
40
    
41
    Returns:
42
    - True if credentials are Private App type
43
    """
44
```
45

46
### Authenticator Creation
47

48
Creates appropriate authenticator instance based on credentials.
49

50
```python { .api }
51
def get_authenticator(self) -> Optional[Union[Oauth2Authenticator, TokenAuthenticator]]:
52
    """
53
    Get authenticator instance based on credentials type.
54
    
55
    Returns:
56
    - Oauth2Authenticator for OAuth credentials
57
    - TokenAuthenticator for Private App credentials
58
    - None for unsupported credential types
59
    """
60
```
61

62
### HTTP Operations
63

64
Performs HTTP requests with authentication, error handling, and retry logic.
65

66
```python { .api }
67
def get(
68
    self, 
69
    url: str, 
70
    params: MutableMapping[str, Any] = None
71
) -> Tuple[Union[MutableMapping[str, Any], List[MutableMapping[str, Any]]], requests.Response]:
72
    """
73
    Perform authenticated GET request with retry logic.
74
    
75
    Parameters:
76
    - url: API endpoint URL (relative to BASE_URL)
77
    - params: Query parameters
78
    
79
    Returns:
80
    - Tuple of (parsed_response_data, raw_response)
81
    """
82

83
def post(
84
    self, 
85
    url: str, 
86
    data: Mapping[str, Any], 
87
    params: MutableMapping[str, Any] = None
88
) -> Tuple[Union[Mapping[str, Any], List[Mapping[str, Any]]], requests.Response]:
89
    """
90
    Perform authenticated POST request.
91
    
92
    Parameters:
93
    - url: API endpoint URL (relative to BASE_URL)
94
    - data: Request body data
95
    - params: Query parameters
96
    
97
    Returns:
98
    - Tuple of (parsed_response_data, raw_response)
99
    """
100
```
101

102
### Custom Object Metadata
103

104
Discovers and processes HubSpot custom object schemas and properties.
105

106
```python { .api }
107
def get_custom_objects_metadata(self) -> Iterable[Tuple[str, str, Mapping[str, Any]]]:
108
    """
109
    Get metadata for all custom objects in the HubSpot account.
110
    
111
    Yields:
112
    - Tuples of (object_name, fully_qualified_name, schema, properties)
113
    """
114

115
def get_properties(self, raw_schema: Mapping[str, Any]) -> Mapping[str, Any]:
116
    """
117
    Extract properties from raw schema definition.
118
    
119
    Parameters:
120
    - raw_schema: Raw schema from HubSpot API
121
    
122
    Returns:
123
    - Dictionary mapping property names to property schemas
124
    """
125

126
def generate_schema(self, properties: Mapping[str, Any]) -> Mapping[str, Any]:
127
    """
128
    Generate JSON schema from properties.
129
    
130
    Parameters:
131
    - properties: Property definitions
132
    
133
    Returns:
134
    - JSON schema for the object
135
    """
136
```
137

138
### Error Handling
139

140
Parses and handles HubSpot API errors with appropriate exception raising.
141

142
```python { .api }
143
@staticmethod
144
def _parse_and_handle_errors(response) -> Union[MutableMapping[str, Any], List[MutableMapping[str, Any]]]:
145
    """
146
    Parse response and handle API errors.
147
    
148
    Parameters:
149
    - response: HTTP response object
150
    
151
    Returns:
152
    - Parsed response data
153
    
154
    Raises:
155
    - HubspotInvalidAuth: For 401/530 authentication errors
156
    - HubspotRateLimited: For 429 rate limit errors
157
    - HubspotTimeout: For 502/503 timeout errors
158
    - HTTPError: For other HTTP errors
159
    """
160
```
161

162
## Usage Examples
163

164
### OAuth Client Setup
165

166
```python
167
from source_hubspot.streams import API
168

169
# OAuth credentials
170
credentials = {
171
    "credentials_title": "OAuth Credentials",
172
    "client_id": "your_client_id",
173
    "client_secret": "your_client_secret",
174
    "refresh_token": "your_refresh_token"
175
}
176

177
# Create API client
178
api = API(credentials)
179

180
# Check authentication type
181
if api.is_oauth2():
182
    print("Using OAuth 2.0 authentication")
183

184
# Make API request
185
data, response = api.get("/crm/v3/objects/contacts", {"limit": 10})
186
print(f"Retrieved {len(data.get('results', []))} contacts")
187
```
188

189
### Private App Client Setup
190

191
```python
192
# Private App credentials
193
credentials = {
194
    "credentials_title": "Private App Credentials",
195
    "access_token": "your_private_app_token"
196
}
197

198
api = API(credentials)
199

200
if api.is_private_app():
201
    print("Using Private App authentication")
202

203
# Get contact data
204
data, response = api.get("/crm/v3/objects/contacts")
205
```
206

207
### Custom Objects Discovery
208

209
```python
210
api = API(credentials)
211

212
# Discover custom objects
213
for name, fqn, schema, properties in api.get_custom_objects_metadata():
214
    print(f"Found custom object: {name}")
215
    print(f"Fully qualified name: {fqn}")
216
    print(f"Properties: {list(properties.keys())}")
217
    print(f"Schema: {schema}")
218
```
219

220
### Error Handling
221

222
```python
223
from source_hubspot.errors import (
224
    HubspotInvalidAuth, HubspotRateLimited, HubspotTimeout
225
)
226

227
try:
228
    data, response = api.get("/crm/v3/objects/contacts")
229
except HubspotInvalidAuth as e:
230
    print(f"Authentication failed: {e}")
231
except HubspotRateLimited as e:
232
    retry_after = e.response.headers.get("Retry-After")
233
    print(f"Rate limited. Retry after {retry_after} seconds")
234
except HubspotTimeout as e:
235
    print(f"Request timeout: {e}")
236
```
237

238
### POST Requests
239

240
```python
241
# Create a new contact
242
contact_data = {
243
    "properties": {
244
        "email": "test@example.com",
245
        "firstname": "John",
246
        "lastname": "Doe"
247
    }
248
}
249

250
try:
251
    data, response = api.post("/crm/v3/objects/contacts", contact_data)
252
    print(f"Created contact with ID: {data['id']}")
253
except Exception as e:
254
    print(f"Failed to create contact: {e}")
255
```
256

257
## Error Response Handling
258

259
The API client automatically handles various HubSpot API error conditions:
260

261
- **400 Bad Request**: Logs warning and raises HTTPError
262
- **401 Unauthorized**: Raises HubspotInvalidAuth
263
- **403 Forbidden**: Logs warning and raises HTTPError  
264
- **429 Too Many Requests**: Raises HubspotRateLimited with retry-after header
265
- **502/503 Server Errors**: Raises HubspotTimeout for retry
266
- **530 Cloudflare DNS Error**: Raises HubspotInvalidAuth (invalid API token format)
267

268
## Constants
269

270
```python { .api }
271
BASE_URL: str = "https://api.hubapi.com"  # HubSpot API base URL
272
USER_AGENT: str = "Airbyte"  # User agent for requests
273
```