0
# Custom Objects
1

2
Support for HubSpot custom objects with dynamic schema generation and custom properties handling. Custom objects allow HubSpot users to create their own data structures beyond the standard CRM objects.
3

4
## Capabilities
5

6
### Custom Object Stream
7

8
Base class for all HubSpot custom object streams with dynamic schema support.
9

10
```python { .api }
11
class CustomObject(CRMSearchStream, ABC):
12
    """
13
    Abstract base class for HubSpot custom object streams.
14
    
15
    Provides dynamic schema generation and property handling
16
    for user-defined custom objects in HubSpot.
17
    """
18
    
19
    def __init__(
20
        self,
21
        entity: str,
22
        schema: Mapping[str, Any],
23
        fully_qualified_name: str,
24
        custom_properties: Mapping[str, Any],
25
        **kwargs
26
    ):
27
        """
28
        Initialize custom object stream.
29
        
30
        Parameters:
31
        - entity: Custom object entity name
32
        - schema: JSON schema for the custom object
33
        - fully_qualified_name: Fully qualified object name from HubSpot
34
        - custom_properties: Custom property definitions
35
        - **kwargs: Additional stream parameters (api, start_date, etc.)
36
        """
37
```
38

39
### Custom Object Discovery
40

41
Methods for discovering and processing custom object metadata from HubSpot.
42

43
```python { .api }
44
# From API class
45
def get_custom_objects_metadata(self) -> Iterable[Tuple[str, str, Mapping[str, Any]]]:
46
    """
47
    Discover custom objects in the HubSpot account.
48
    
49
    Yields:
50
    - Tuples of (entity_name, fully_qualified_name, schema, properties)
51
    """
52

53
def get_properties(self, raw_schema: Mapping[str, Any]) -> Mapping[str, Any]:
54
    """
55
    Extract properties from custom object schema.
56
    
57
    Parameters:
58
    - raw_schema: Raw schema definition from HubSpot API
59
    
60
    Returns:
61
    - Dictionary mapping property names to property definitions
62
    """
63

64
def generate_schema(self, properties: Mapping[str, Any]) -> Mapping[str, Any]:
65
    """
66
    Generate JSON schema from custom properties.
67
    
68
    Parameters:
69
    - properties: Property definitions
70
    
71
    Returns:
72
    - JSON schema for the custom object
73
    """
74
```
75

76
### Custom Object Stream Generation
77

78
Methods in SourceHubspot for creating custom object stream instances.
79

80
```python { .api }
81
def get_custom_object_streams(self, api: API, common_params: Mapping[str, Any]) -> Generator[CustomObject, None, None]:
82
    """
83
    Generate custom object stream instances.
84
    
85
    Parameters:
86
    - api: API client instance
87
    - common_params: Common stream parameters
88
    
89
    Yields:
90
    - CustomObject stream instances for each discovered custom object
91
    """
92
```
93

94
## Usage Examples
95

96
### Discovering Custom Objects
97

98
```python
99
from source_hubspot.streams import API
100
from source_hubspot import SourceHubspot
101

102
# Setup API client
103
api = API(credentials)
104

105
# Discover custom objects
106
print("Discovering custom objects...")
107
for entity, fqn, schema, properties in api.get_custom_objects_metadata():
108
    print(f"\nCustom Object: {entity}")
109
    print(f"Fully Qualified Name: {fqn}")
110
    print(f"Properties: {list(properties.keys())}")
111
    print(f"Schema type: {schema.get('type', 'unknown')}")
112
    
113
    # Show property details
114
    for prop_name, prop_def in properties.items():
115
        prop_type = prop_def.get('type', 'unknown')
116
        print(f"  - {prop_name}: {prop_type}")
117
```
118

119
### Working with Custom Object Streams
120

121
```python
122
from source_hubspot import SourceHubspot
123

124
# Create source with custom objects enabled
125
source = SourceHubspot(catalog=None, config=config, state=None)
126

127
# Get all streams including custom objects
128
all_streams = source.streams(config)
129

130
# Filter for custom object streams
131
custom_streams = []
132
for stream in all_streams:
133
    if hasattr(stream, 'entity') and hasattr(stream, 'fully_qualified_name'):
134
        custom_streams.append(stream)
135

136
print(f"Found {len(custom_streams)} custom object streams:")
137
for stream in custom_streams:
138
    print(f"  - {stream.name} ({stream.entity})")
139
```
140

141
### Reading Custom Object Data
142

143
```python
144
# Assuming we have a custom object stream for "Vehicle" objects
145
for stream in custom_streams:
146
    if stream.entity == "2-123456":  # Custom object ID example
147
        print(f"\nReading data from custom object: {stream.name}")
148
        
149
        record_count = 0
150
        for record in stream.read_records(sync_mode="full_refresh"):
151
            record_count += 1
152
            properties = record.get('properties', {})
153
            
154
            # Display first few records
155
            if record_count <= 3:
156
                print(f"Record {record_count}:")
157
                for prop_name, prop_value in properties.items():
158
                    print(f"  {prop_name}: {prop_value}")
159
        
160
        print(f"Total records: {record_count}")
161
        break
162
```
163

164
### Custom Object Schema Analysis
165

166
```python
167
# Analyze custom object schemas in detail
168
api = API(credentials)
169

170
for entity, fqn, schema, properties in api.get_custom_objects_metadata():
171
    print(f"\n=== Custom Object: {entity} ===")
172
    print(f"Fully Qualified Name: {fqn}")
173
    
174
    # Analyze schema structure
175
    if 'properties' in schema:
176
        schema_props = schema['properties']
177
        print(f"Schema properties: {len(schema_props)}")
178
        
179
        # Categorize property types
180
        type_counts = {}
181
        for prop_name, prop_schema in schema_props.items():
182
            prop_type = prop_schema.get('type', 'unknown')
183
            if prop_type not in type_counts:
184
                type_counts[prop_type] = 0
185
            type_counts[prop_type] += 1
186
        
187
        print("Property type distribution:")
188
        for prop_type, count in sorted(type_counts.items()):
189
            print(f"  {prop_type}: {count}")
190
    
191
    # Analyze custom properties
192
    print(f"Custom properties: {len(properties)}")
193
    required_props = []
194
    optional_props = []
195
    
196
    for prop_name, prop_def in properties.items():
197
        if prop_def.get('required', False):
198
            required_props.append(prop_name)
199
        else:
200
            optional_props.append(prop_name)
201
    
202
    if required_props:
203
        print(f"Required properties: {', '.join(required_props)}")
204
    if optional_props:
205
        print(f"Optional properties: {len(optional_props)} total")
206
```
207

208
### Custom Object Stream Configuration
209

210
```python
211
# Manually create a custom object stream
212
from source_hubspot.streams import CustomObject
213

214
# This would typically be done automatically by SourceHubspot
215
custom_stream = CustomObject(
216
    entity="2-123456",  # HubSpot custom object ID
217
    schema={
218
        "type": "object",
219
        "properties": {
220
            "vehicle_make": {"type": "string"},
221
            "vehicle_model": {"type": "string"},
222
            "year": {"type": "integer"},
223
            "price": {"type": "number"}
224
        }
225
    },
226
    fully_qualified_name="p123456_vehicle",
227
    custom_properties={
228
        "vehicle_make": {"type": "string", "label": "Vehicle Make"},
229
        "vehicle_model": {"type": "string", "label": "Vehicle Model"},
230
        "year": {"type": "integer", "label": "Year"},
231
        "price": {"type": "number", "label": "Price"}
232
    },
233
    api=api,
234
    start_date="2023-01-01T00:00:00Z",
235
    credentials=credentials
236
)
237

238
# Use the custom stream
239
for record in custom_stream.read_records(sync_mode="full_refresh"):
240
    properties = record['properties']
241
    print(f"Vehicle: {properties.get('vehicle_make')} {properties.get('vehicle_model')} ({properties.get('year')})")
242
    print(f"Price: ${properties.get('price', 0):,.2f}")
243
```
244

245
### Custom Object Web Analytics (Experimental)
246

247
```python
248
# If experimental streams are enabled, custom objects get web analytics streams too
249
config_with_experimental = {
250
    **config,
251
    "enable_experimental_streams": True
252
}
253

254
source = SourceHubspot(catalog=None, config=config_with_experimental, state=None)
255
streams = source.streams(config_with_experimental)
256

257
# Find web analytics streams for custom objects
258
custom_web_analytics = [
259
    s for s in streams 
260
    if "WebAnalytics" in s.__class__.__name__ and hasattr(s, 'parent')
261
]
262

263
print(f"Found {len(custom_web_analytics)} custom object web analytics streams")
264
for stream in custom_web_analytics:
265
    print(f"  - {stream.name}")
266
```
267

268
## Custom Object Properties
269

270
### Property Types
271

272
HubSpot custom objects support various property types:
273

274
```python { .api }
275
# Common custom property types
276
CUSTOM_FIELD_TYPE_TO_VALUE = {
277
    bool: "boolean",
278
    str: "string", 
279
    float: "number",
280
    int: "integer"
281
}
282

283
# Schema type conversions
284
KNOWN_CONVERTIBLE_SCHEMA_TYPES = {
285
    "bool": ("boolean", None),
286
    "enumeration": ("string", None),
287
    "date": ("string", "date"),
288
    "date-time": ("string", "date-time"),
289
    "datetime": ("string", "date-time"),
290
    "json": ("string", None),
291
    "phone_number": ("string", None)
292
}
293
```
294

295
### Property Schema Structure
296

297
```python { .api }
298
# Example custom property definition
299
{
300
    "name": "property_name",
301
    "label": "Display Label",
302
    "type": "string",  # or "number", "boolean", "datetime", etc.
303
    "fieldType": "text",  # HubSpot field type
304
    "required": False,
305
    "options": [],  # For enumeration types
306
    "description": "Property description"
307
}
308
```
309

310
## OAuth Scopes
311

312
Custom object streams require specific OAuth scopes:
313

314
- **Default Custom Objects**: `crm.schemas.custom.read`, `crm.objects.custom.read`
315
- **Custom Object Creation**: `crm.schemas.custom.write`, `crm.objects.custom.write`
316
- **Custom Object Properties**: Additional schema read/write scopes may be required
317

318
The connector automatically detects available custom objects based on granted scopes and includes only those that are accessible to the authenticated user.
319

320
## Limitations
321

322
- Custom object availability depends on HubSpot plan and permissions
323
- Property types must be supported by HubSpot's custom object system
324
- Schema changes to custom objects may require connector reconfiguration
325
- Some advanced custom object features may not be available through the API