Stevan/onspring-api-sdk-python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: rewrite SDK from camelCase/requests to snake_case/httpx/Pydantic v2

Browse Source 
- Replace OnspringApiSdk package with onspring_api_sdk package
- Add sync (OnspringClient) and async (AsyncOnspringClient) clients
- Migrate from requests to httpx for HTTP transport
- Migrate from raw dataclasses to Pydantic v2 models with alias support
- Add shared response handler module (_responses.py) to reduce duplication
- Add typed exceptions (OnspringError, OnspringAuthenticationError, etc.)
- Preserve full API coverage for apps, fields, files, lists, records, and reports
This commit is contained in:
Stevan Freeborn
2026-06-26 17:42:56 -05:00
parent bc0aa9296b
commit b976959c30
22 changed files with 2339 additions and 3515 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/OnspringApiSDK.Tests/Tests_OnspringClient.py
-219
View File
@@ -1,219 +0,0 @@
import sys
import requests_mock
sys.path.append('src')
from OnspringApiSdk.OnspringClient import OnspringClient
@requests_mock.Mocker(kw='mock')
class TestOnspringClient(object):
test_url = 'https://test.com'
test_apiKey = 'apiKey'
client = OnspringClient(test_url, test_apiKey)
def test_GetFieldsByAppId_WhenFieldContainsListValues_ItShouldReturnThoseValues(self, **kwargs):
mockResponse = {
"pageNumber": 1,
"pageSize": 2,
"totalPages": 1,
"totalRecords": 2,
"items": [
{
"multiplicity": "SingleSelect",
"listId": 638,
"values": [
{
"id": "2c1af5b1-0f90-4378-b9a5-8b7e22f2bc84",
"name": "list_value_1",
"sortOrder": 1,
"numericValue": 1,
"color": "#008e8e"
},
{
"id": "0421e502-7f76-480a-9311-363aca3560bc",
"name": "list_value_2",
"sortOrder": 2,
"numericValue": 2,
"color": "#a186be"
},
{
"id": "285b91c1-5800-47cb-a030-8cf7cdd7cdf1",
"name": "updated_list_value_1676840661138}",
"sortOrder": 3,
"numericValue": 1,
"color": "#000000"
}
],
"id": 4801,
"appId": 130,
"name": "single_select_list_field",
"type": "List",
"status": "Enabled",
"isRequired": True,
"isUnique": False
},
{
"outputType": "ListValue",
"values": [
{
"id": "b235afb2-b786-4c87-bce9-fbd700e246c1",
"name": "list_value_1",
"sortOrder": 1,
"numericValue": 1,
"color": "#6dcff6"
},
{
"id": "5cd7cd55-d6a6-40e0-a560-8aa407c13210",
"name": "list_value_2",
"sortOrder": 2,
"numericValue": 2,
"color": "#8e468e"
}
],
"id": 4815,
"appId": 130,
"name": "list_formula_field",
"type": "Formula",
"status": "Enabled",
"isRequired": False,
"isUnique": False
},
]
}
kwargs['mock'].get(self.test_url + '/Fields/appId/1', json=mockResponse)
response = self.client.GetFieldsByAppId(1)
field_4801 = response.data.fields[0]
field_4815 = response.data.fields[1]
assert field_4801.id == 4801, "Field id should be 4801"
assert field_4801.listId == 638, "Field list id should be 638"
assert field_4801.multiplicity == "SingleSelect", "Field multiplicity should be SingleSelect"
assert field_4801.values is not None, "Field list values should not be None"
field_4801_value = field_4801.values[0]
assert field_4801_value.id == "2c1af5b1-0f90-4378-b9a5-8b7e22f2bc84", "Field value id should be 2c1af5b1-0f90-4378-b9a5-8b7e22f2bc84"
assert field_4801_value.name == "list_value_1", "Field value name should be list_value_1"
assert field_4801_value.sortOrder == 1, "Field value sort order should be 1"
assert field_4801_value.numericValue == 1, "Field value numeric value should be 1"
assert field_4801_value.color == "#008e8e", "Field value color should be #008e8e"
assert field_4815.id == 4815, "Field id should be 4815"
assert field_4815.listId is None, "Field list id should be None"
assert field_4815.multiplicity is None, "Field multiplicity should be None"
assert field_4815.values is not None, "Field list values should not be None"
assert field_4815.outputType == 'ListValue', 'Field output type should be ListValue'
field_4815_value = field_4815.values[0]
assert field_4815_value.id == "b235afb2-b786-4c87-bce9-fbd700e246c1", "Field value id should be b235afb2-b786-4c87-bce9-fbd700e246c1"
assert field_4815_value.name == "list_value_1", "Field value name should be list_value_1"
assert field_4815_value.sortOrder == 1, "Field value sort order should be 1"
assert field_4815_value.numericValue == 1, "Field value numeric value should be 1"
assert field_4815_value.color == "#6dcff6", "Field value color should be #6dcff6"
def test_GetFieldsByIds_WhenFieldContainsListValues_ItShouldReturnThoseValues(self, **kwargs):
mockResponse = {
"count": 2,
"items": [
{
"multiplicity": "SingleSelect",
"listId": 638,
"values": [
{
"id": "2c1af5b1-0f90-4378-b9a5-8b7e22f2bc84",
"name": "list_value_1",
"sortOrder": 1,
"numericValue": 1,
"color": "#008e8e"
},
{
"id": "0421e502-7f76-480a-9311-363aca3560bc",
"name": "list_value_2",
"sortOrder": 2,
"numericValue": 2,
"color": "#a186be"
},
{
"id": "285b91c1-5800-47cb-a030-8cf7cdd7cdf1",
"name": "updated_list_value_1676840661138}",
"sortOrder": 3,
"numericValue": 1,
"color": "#000000"
}
],
"id": 4801,
"appId": 130,
"name": "single_select_list_field",
"type": "List",
"status": "Enabled",
"isRequired": True,
"isUnique": False
},
{
"outputType": "ListValue",
"values": [
{
"id": "b235afb2-b786-4c87-bce9-fbd700e246c1",
"name": "list_value_1",
"sortOrder": 1,
"numericValue": 1,
"color": "#6dcff6"
},
{
"id": "5cd7cd55-d6a6-40e0-a560-8aa407c13210",
"name": "list_value_2",
"sortOrder": 2,
"numericValue": 2,
"color": "#8e468e"
}
],
"id": 4815,
"appId": 130,
"name": "list_formula_field",
"type": "Formula",
"status": "Enabled",
"isRequired": False,
"isUnique": False
}
]
}
kwargs['mock'].post(self.test_url + '/Fields/batch-get', json=mockResponse)
response = self.client.GetFieldsByIds([4801, 4815])
field_4801 = response.data.fields[0]
field_4815 = response.data.fields[1]
assert field_4801.id == 4801, "Field id should be 4801"
assert field_4801.listId == 638, "Field list id should be 638"
assert field_4801.multiplicity == "SingleSelect", "Field multiplicity should be SingleSelect"
assert field_4801.values is not None, "Field list values should not be None"
field_4801_value = field_4801.values[0]
assert field_4801_value.id == "2c1af5b1-0f90-4378-b9a5-8b7e22f2bc84", "Field value id should be 2c1af5b1-0f90-4378-b9a5-8b7e22f2bc84"
assert field_4801_value.name == "list_value_1", "Field value name should be list_value_1"
assert field_4801_value.sortOrder == 1, "Field value sort order should be 1"
assert field_4801_value.numericValue == 1, "Field value numeric value should be 1"
assert field_4801_value.color == "#008e8e", "Field value color should be #008e8e"
assert field_4815.id == 4815, "Field id should be 4815"
assert field_4815.listId is None, "Field list id should be None"
assert field_4815.multiplicity is None, "Field multiplicity should be None"
assert field_4815.values is not None, "Field list values should not be None"
assert field_4815.outputType == 'ListValue', 'Field output type should be ListValue'
field_4815_value = field_4815.values[0]
assert field_4815_value.id == "b235afb2-b786-4c87-bce9-fbd700e246c1", "Field value id should be b235afb2-b786-4c87-bce9-fbd700e246c1"
assert field_4815_value.name == "list_value_1", "Field value name should be list_value_1"
assert field_4815_value.sortOrder == 1, "Field value sort order should be 1"
assert field_4815_value.numericValue == 1, "Field value numeric value should be 1"
assert field_4815_value.color == "#6dcff6", "Field value color should be #6dcff6"
src/OnspringApiSdk/Endpoints.py
-323
View File
@@ -1,323 +0,0 @@
import uuid
# connectivity endpoints
def GetPingEndpoint(baseUrl: str) -> str:
"""
Returns the ping endpoint.
Args:
baseUrl (`str`): The base url for the api.
Returns:
The ping endpoint as a string.
"""
return f'{baseUrl}/Ping'
# app endpoints
def GetAppsEndpoint(baseUrl: str) -> str:
"""
Returns the get apps endpoint.
Args:
baseUrl (`str`): The base url for the api.
Returns:
The get apps endpoint as a string.
"""
return f'{baseUrl}/Apps'
def GetAppByIdEndpoint(baseUrl: str, appId: int) -> str:
"""
Returns the get app by id endpoint.
Args:
baseUrl (`str`): The base url for the api.
appId (`int`): The id of the app being requested.
Returns:
The get app by id endpoint as a string.
"""
return f'{baseUrl}/Apps/id/{appId}'
def GetAppsByIdsEndpoint(baseUrl: str) -> str:
"""
Returns the get apps by ids endpoint.
Args:
baseUrl (`str`): The base url for the api.
Returns:
The get apps by ids endpoint as a string.
"""
return f'{baseUrl}/Apps/batch-get'
# field endpoints
def GetFieldByIdEndpoint(baseUrl: str, fieldId: int) -> str:
"""
Returns the get field by id endpoint.
Args:
baseUrl (`str`): The base url for the api.
fieldId (`int`): The id of the field being requested.
Returns:
The get field by id endpoint as a string.
"""
return f'{baseUrl}/Fields/id/{fieldId}'
def GetFieldsByIdsEndpoint(baseUrl: str) -> str:
"""
Returns the get fields by ids endpoint.
Args:
baseUrl (`str`): The base url for the api.
Returns:
The get fields by ids endpoint as a string.
"""
return f'{baseUrl}/Fields/batch-get'
def GetFieldsByAppIdEndpoint(baseUrl: str, appId: int) -> str:
"""
Returns the get fields by app id endpoint.
Args:
baseUrl (`str`): The base url for the api.
appId (`int`): The id of the app whose fields are being requested.
Returns:
The get fields by app id endpoint as a string.
"""
return f'{baseUrl}/Fields/appId/{appId}'
# file endpoints
def GetFileInfoByIdEndpoint(baseUrl: str, recordId: int, fieldId: int, fileId: int) -> str:
"""
Returns the get file info by its id endpoint.
Args:
baseUrl (`str`): The base url for the api.
recordId (`int`): The id for the record where the file resides.
fieldId (`int`): The id for the field in the record where the file is held.
fileId ('int'): The id of the file.
Returns:
The get file info by its id endpoint as a string.
"""
return f'{baseUrl}/Files/recordId/{recordId}/fieldId/{fieldId}/fileId/{fileId}'
def DeleteFileByIdEndpoint(baseUrl: str, recordId: int, fieldId: int, fileId: int) -> str:
"""
Returns the delete file by its id endpoint.
Args:
baseUrl (`str`): The base url for the api.
recordId (`int`): The id for the record where the file resides.
fieldId (`int`): The id for the field in the record where the file is held.
fileId ('int'): The id of the file.
Returns:
The delete file by its id endpoint as a string.
"""
return f'{baseUrl}/Files/recordId/{recordId}/fieldId/{fieldId}/fileId/{fileId}'
def GetFileByIdEndpoint(baseUrl: str, recordId: int, fieldId: int, fileId: int) -> str:
"""
Returns the get file by its id endpoint.
Args:
baseUrl (`str`): The base url for the api.
recordId (`int`): The id for the record where the file resides.
fieldId (`int`): The id for the field in the record where the file is held.
fileId ('int'): The id of the file.
Returns:
The get file by its id endpoint as a string.
"""
return f'{baseUrl}/Files/recordId/{recordId}/fieldId/{fieldId}/fileId/{fileId}/file'
def SaveFileEndpoint(baseUrl: str) -> str:
"""
Returns the save file endpoint.
Args:
baseUrl (`str`): The base url for the api.
Returns:
The save file endpoint as a string.
"""
return f'{baseUrl}/Files'
# list endpoints
def AddOrUpdateListItemEndpoint(baseUrl, listId: int) -> str:
"""
Returns the add or update list item endpoint.
Args:
baseUrl (`str`): The base url for the api.
listId (`int`): The id of the list for which the item belongs when being updated or being added to.
Returns:
The add or update list item endpoint as a string.
"""
return f'{baseUrl}/Lists/id/{listId}/items'
def DeleteListItemEndpoint(baseUrl: str, listId: int, itemId: uuid) -> str:
"""
Returns the delete list item endpoint.
Args:
baseUrl (`str`): The base url for the api.
listId (`int`): The id of the list for which the list item belongs.
itemId (`int`): The id list item.
Returns:
The delete list item endpoint as a string.
"""
return f'{baseUrl}/Lists/id/{listId}/itemId/{itemId}'
# record endpoints
def GetRecordsByAppIdEndpoint(baseUrl, appId: int) -> str:
"""
Returns the get records by app id endpoint.
Args:
baseUrl (`str`): The base url for the api.
appId (`int`): The id of the app where the records reside.
Returns:
The get records by app id endpoint as a string.
"""
return f'{baseUrl}/Records/appId/{appId}'
def GetRecordByIdEndpoint(baseUrl, appId: int, recordId: int) -> str:
"""
Returns the get record by id endpoint.
Args:
baseUrl (`str`): The base url for the api.
appId (`int`): The id of the app where the record resides.
recordId (`int`): The id of the record.
Returns:
The get record by id endpoint as a string.
"""
return f'{baseUrl}/Records/appId/{appId}/recordId/{recordId}'
def DeleteRecordByIdEndpoint(baseUrl, appId: int, recordId: int) -> str:
"""
Returns the delete record by id endpoint.
Args:
baseUrl (`str`): The base url for the api.
appId (`int`): The id of the app where the record resides.
recordId (`int`): The id of the record.
Returns:
The delete record by id endpoint as a string.
"""
return f'{baseUrl}/Records/appId/{appId}/recordId/{recordId}'
def GetRecordsByIdsEndpoint(baseUrl) -> str:
"""
Returns the get records by ids endpoint.
Args:
baseUrl (`str`): The base url for the api.
Returns:
The get records by ids endpoint as a string.
"""
return f'{baseUrl}/Records/batch-get'
def QueryRecordsEndpoint(baseUrl) -> str:
"""
Returns the query records endpoint.
Args:
baseUrl (`str`): The base url for the api.
Returns:
The query records endpoint as a string.
"""
return f'{baseUrl}/Records/Query'
def AddOrUpdateRecordEndpoint(baseUrl) -> str:
"""
Returns the add or update record endpoint.
Args:
baseUrl (`str`): The base url for the api.
Returns:
The add or update record endpoint as a string.
"""
return f'{baseUrl}/Records'
def DeleteRecordsByIds(baseUrl) -> str:
"""
Returns the delete records by ids endpoint.
Args:
baseUrl (`str`): The base url for the api.
Returns:
The delete records by ids endpoint as a string.
"""
return f'{baseUrl}/Records/batch-delete'
# report endpoints
def GetReportByIdEndpoint(baseUrl, reportId: int) -> str:
"""
Returns the get report by id endpoint.
Args:
baseUrl (`str`): The base url for the api.
reportId (`int`): The id of the report.
Returns:
The get report by id endpoint as a string.
"""
return f'{baseUrl}/Reports/id/{reportId}'
def GetReportsByAppIdEndpoint(baseUrl, appId: int) -> str:
"""
Returns the get reports by app id endpoint.
Args:
baseUrl (`str`): The base url for the api.
appId (`int`): The id of the app where the reports reside.
Returns:
The get reports by app id endpoint as a string.
"""
return f'{baseUrl}/Reports/appId/{appId}'
src/OnspringApiSdk/Enums.py
-54
View File
@@ -1,54 +0,0 @@
from enum import Enum
class DataFormat(Enum):
"""
The possible data format types for record field values.
"""
Raw:int = 0
Formatted:int = 1
class ReportDataType(Enum):
"""
The possible report data types for reports.
"""
ReportData:int = 0
ChartData:int = 1
class ResultValueType(Enum):
"""
The possible types for record field values.
"""
String:int = 0
Integer:int = 1
Decimal:int = 2
Date:int = 3
TimeSpan:int = 4
Guid:int = 5
StringList:int = 6
IntegerList:int = 7
GuidList:int = 8
AttachmentList:int = 9
ScoringGroupList:int = 10
FileList:int = 11
class Increment(Enum):
"""
The possible values for the increment property of timespan data in an Onspring timespan field.
"""
Seconds:str = "Second(s)"
Minutes:str = "Minute(s)"
Hours:str = "Hour(s)"
Days:str = "Day(s)"
Weeks:str = "Week(s)"
Months:str = "Month(s)"
Years:str = "Year(s)"
class Recurrence(Enum):
"""
The possible values for the recurrence property of timespan data in an Onspring timespan field.
"""
Empty:str = "None"
EndByDate:str = "EndByDate"
EndAfterOccurrences:str = 'EndAfterOccurrences'
src/OnspringApiSdk/Helpers.py
-13
View File
@@ -1,13 +0,0 @@
from datetime import datetime
from OnspringApiSdk.Enums import *
def parseDate(date: str) -> datetime:
if date==None:
return None
for format in ["%Y-%m-%dT%H:%M:%S.%fZ","%Y-%m-%dT%H:%M:%SZ"]:
try:
return datetime.strptime(date, format)
except ValueError:
pass
src/OnspringApiSdk/Models.py
-1221
View File
File diff suppressed because it is too large Load Diff
src/OnspringApiSdk/OnspringClient.py
-1685
View File
File diff suppressed because it is too large Load Diff
src/OnspringApiSdk/__init__.py
View File
src/onspring_api_sdk/__init__.py
+22
View File
@@ -0,0 +1,22 @@
"""Onspring API SDK.
Provides sync and async clients for interacting with the Onspring API v2.
"""
from onspring_api_sdk.async_client import AsyncOnspringClient
from onspring_api_sdk.client import OnspringClient
from onspring_api_sdk.errors import (
OnspringAuthenticationError,
OnspringError,
OnspringNotFoundError,
OnspringRateLimitError,
)
__all__ = [
"OnspringClient",
"AsyncOnspringClient",
"OnspringError",
"OnspringAuthenticationError",
"OnspringNotFoundError",
"OnspringRateLimitError",
]
src/onspring_api_sdk/_responses.py
+734
View File
@@ -0,0 +1,734 @@
"""Shared response handlers for Onspring API endpoints."""
import re
import httpx
from onspring_api_sdk.errors import _get_error_message
from onspring_api_sdk.models import (
AddOrUpdateListItemResponse,
AddOrUpdateRecordResponse,
ApiResponse,
App,
File,
FileInfo,
GetAppByIdResponse,
GetAppsByIdsResponse,
GetAppsResponse,
GetBatchRecordsResponse,
GetFieldByIdResponse,
GetFieldsByAppIdResponse,
GetFieldsByIdsResponse,
GetFileByIdResponse,
GetFileInfoByIdResponse,
GetRecordsResponse,
GetReportByIdResponse,
GetReportsByAppIdResponse,
OnspringField,
Record,
SaveFileResponse,
)
def handle_get_apps_response(response: httpx.Response) -> ApiResponse[GetAppsResponse]:
match response.status_code:
case 400:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Invalid paging information",
raw_response=response,
)
case 401:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Unauthorized request",
raw_response=response,
)
case 200:
return ApiResponse(
status_code=response.status_code,
data=GetAppsResponse.model_validate(response.json()),
raw_response=response,
)
case _:
return ApiResponse(status_code=response.status_code, is_successful=False, raw_response=response)
def handle_get_field_by_id_response(response: httpx.Response) -> ApiResponse[GetFieldByIdResponse]:
match response.status_code:
case 401:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Unauthorized request",
raw_response=response,
)
case 403:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Client does not have read access to the field",
raw_response=response,
)
case 404:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Field could not be found",
raw_response=response,
)
case 200:
return ApiResponse(
status_code=response.status_code,
data=GetFieldByIdResponse(field=OnspringField.model_validate(response.json())),
raw_response=response,
)
case _:
return ApiResponse(status_code=response.status_code, is_successful=False, raw_response=response)
def handle_get_fields_by_ids_response(response: httpx.Response) -> ApiResponse[GetFieldsByIdsResponse]:
match response.status_code:
case 401:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Unauthorized request",
raw_response=response,
)
case 403:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Client does not have read access to the field(s)",
raw_response=response,
)
case 404:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Field(s) could not be found",
raw_response=response,
)
case 200:
return ApiResponse(
status_code=response.status_code,
data=GetFieldsByIdsResponse.model_validate(response.json()),
raw_response=response,
)
case _:
return ApiResponse(status_code=response.status_code, is_successful=False, raw_response=response)
def handle_get_fields_by_app_id_response(response: httpx.Response) -> ApiResponse[GetFieldsByAppIdResponse]:
match response.status_code:
case 400:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Invalid paging information",
raw_response=response,
)
case 401:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Unauthorized request",
raw_response=response,
)
case 200:
return ApiResponse(
status_code=response.status_code,
data=GetFieldsByAppIdResponse.model_validate(response.json()),
raw_response=response,
)
case _:
return ApiResponse(status_code=response.status_code, is_successful=False, raw_response=response)
def handle_get_apps_by_ids_response(response: httpx.Response) -> ApiResponse[GetAppsByIdsResponse]:
match response.status_code:
case 401:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Unauthorized request",
raw_response=response,
)
case 403:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Client does not have read access to the app",
raw_response=response,
)
case 200:
return ApiResponse(
status_code=response.status_code,
data=GetAppsByIdsResponse.model_validate(response.json()),
raw_response=response,
)
case _:
return ApiResponse(status_code=response.status_code, is_successful=False, raw_response=response)
def handle_get_app_by_id_response(response: httpx.Response) -> ApiResponse[GetAppByIdResponse]:
match response.status_code:
case 401:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Unauthorized request",
raw_response=response,
)
case 403:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Client does not have read access to the app",
raw_response=response,
)
case 404:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="App could not be found",
raw_response=response,
)
case 200:
return ApiResponse(
status_code=response.status_code,
data=GetAppByIdResponse(app=App.model_validate(response.json())),
raw_response=response,
)
case _:
return ApiResponse(status_code=response.status_code, is_successful=False, raw_response=response)
def handle_get_file_info_by_id_response(response: httpx.Response) -> ApiResponse[GetFileInfoByIdResponse]:
match response.status_code:
case 400:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Request is invalid based on underlying data",
raw_response=response,
)
case 401:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Unauthorized request",
raw_response=response,
)
case 403:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Client does not have read access to the file",
raw_response=response,
)
case 404:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="File could not be found",
raw_response=response,
)
case 200:
return ApiResponse(
status_code=response.status_code,
data=GetFileInfoByIdResponse(file_info=FileInfo.model_validate(response.json())),
raw_response=response,
)
case _:
return ApiResponse(status_code=response.status_code, is_successful=False, raw_response=response)
def handle_delete_file_by_id_response(response: httpx.Response) -> ApiResponse[None]:
match response.status_code:
case 400:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Request is invalid based on underlying data",
raw_response=response,
)
case 401:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Unauthorized request",
raw_response=response,
)
case 403 | 404:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message=_get_error_message(response),
raw_response=response,
)
case 500:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="File could not be deleted due to internal error",
raw_response=response,
)
case 204:
return ApiResponse(
status_code=response.status_code, message="File deleted successfully", raw_response=response
)
case _:
return ApiResponse(status_code=response.status_code, is_successful=False, raw_response=response)
def handle_get_file_by_id_response(response: httpx.Response) -> ApiResponse[GetFileByIdResponse]:
match response.status_code:
case 400:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Request is invalid based on underlying data",
raw_response=response,
)
case 401:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Unauthorized request",
raw_response=response,
)
case 403 | 404:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message=_get_error_message(response),
raw_response=response,
)
case 200:
headers = response.headers
content_disposition = headers.get("content-disposition", "")
match = re.search(r"filename=(.*?)(?:;|$)", content_disposition)
file_name = match.group(1).strip("'\"") if match else "OnspringFile"
file = File(
name=file_name,
contentType=headers.get("content-type", ""),
contentLength=int(headers.get("content-length", 0)),
content=response.content,
)
return ApiResponse(
status_code=response.status_code, data=GetFileByIdResponse(file=file), raw_response=response
)
case _:
return ApiResponse(status_code=response.status_code, is_successful=False, raw_response=response)
def handle_save_file_response(response: httpx.Response) -> ApiResponse[SaveFileResponse]:
match response.status_code:
case 400:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Request is invalid based on underlying data",
raw_response=response,
)
case 401:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Unauthorized request",
raw_response=response,
)
case 403 | 404 | 500:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message=_get_error_message(response),
raw_response=response,
)
case 201:
return ApiResponse(
status_code=response.status_code,
data=SaveFileResponse.model_validate(response.json()),
raw_response=response,
)
case _:
return ApiResponse(status_code=response.status_code, is_successful=False, raw_response=response)
def handle_add_or_update_list_item_response(
response: httpx.Response,
) -> ApiResponse[AddOrUpdateListItemResponse]:
match response.status_code:
case 401:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Unauthorized request",
raw_response=response,
)
case 403 | 404:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message=_get_error_message(response),
raw_response=response,
)
case 201:
return ApiResponse(
status_code=response.status_code,
data=AddOrUpdateListItemResponse.model_validate(response.json()),
message="New list value successfully added",
raw_response=response,
)
case 200:
return ApiResponse(
status_code=response.status_code,
data=AddOrUpdateListItemResponse.model_validate(response.json()),
message="Existing list value successfully updated",
raw_response=response,
)
case _:
return ApiResponse(status_code=response.status_code, is_successful=False, raw_response=response)
def handle_delete_list_item_response(response: httpx.Response) -> ApiResponse[None]:
match response.status_code:
case 401:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Unauthorized request",
raw_response=response,
)
case 403:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message=_get_error_message(response),
raw_response=response,
)
case 404:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="List/item could not be found",
raw_response=response,
)
case 204:
return ApiResponse(
status_code=response.status_code, message="Item deleted successfully", raw_response=response
)
case _:
return ApiResponse(status_code=response.status_code, is_successful=False, raw_response=response)
def handle_get_records_by_app_id_response(response: httpx.Response) -> ApiResponse[GetRecordsResponse]:
match response.status_code:
case 400:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Invalid paging information/size of the data requested was too large.",
raw_response=response,
)
case 401:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Unauthorized request",
raw_response=response,
)
case 403:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message=_get_error_message(response),
raw_response=response,
)
case 200:
return ApiResponse(
status_code=response.status_code,
data=GetRecordsResponse.model_validate(response.json()),
raw_response=response,
)
case _:
return ApiResponse(status_code=response.status_code, is_successful=False, raw_response=response)
def handle_get_record_by_id_response(response: httpx.Response) -> ApiResponse[Record]:
match response.status_code:
case 401:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Unauthorized request",
raw_response=response,
)
case 403:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message=_get_error_message(response),
raw_response=response,
)
case 404:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Record could not be found",
raw_response=response,
)
case 200:
return ApiResponse(
status_code=response.status_code, data=Record.model_validate(response.json()), raw_response=response
)
case _:
return ApiResponse(status_code=response.status_code, is_successful=False, raw_response=response)
def handle_delete_record_by_id_response(response: httpx.Response) -> ApiResponse[None]:
match response.status_code:
case 401:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Unauthorized request",
raw_response=response,
)
case 403:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message=_get_error_message(response),
raw_response=response,
)
case 404:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Record could not be found",
raw_response=response,
)
case 204:
return ApiResponse(
status_code=response.status_code, message="Record deleted successfully", raw_response=response
)
case _:
return ApiResponse(status_code=response.status_code, is_successful=False, raw_response=response)
def handle_get_records_by_ids_response(response: httpx.Response) -> ApiResponse[GetBatchRecordsResponse]:
match response.status_code:
case 400:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Batch request is invalid/size of the data requested was too large.",
raw_response=response,
)
case 401:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Unauthorized request",
raw_response=response,
)
case 403:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message=_get_error_message(response),
raw_response=response,
)
case 200:
return ApiResponse(
status_code=response.status_code,
data=GetBatchRecordsResponse.model_validate(response.json()),
raw_response=response,
)
case _:
return ApiResponse(status_code=response.status_code, is_successful=False, raw_response=response)
def handle_query_records_response(response: httpx.Response) -> ApiResponse[GetRecordsResponse]:
match response.status_code:
case 400:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Query request is invalid/size of the data requested was too large.",
raw_response=response,
)
case 401:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Unauthorized request",
raw_response=response,
)
case 403:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message=_get_error_message(response),
raw_response=response,
)
case 200:
return ApiResponse(
status_code=response.status_code,
data=GetRecordsResponse.model_validate(response.json()),
raw_response=response,
)
case _:
return ApiResponse(status_code=response.status_code, is_successful=False, raw_response=response)
def handle_add_or_update_record_response(response: httpx.Response) -> ApiResponse[AddOrUpdateRecordResponse]:
match response.status_code:
case 400:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Request data is invalid",
raw_response=response,
)
case 401:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Unauthorized request",
raw_response=response,
)
case 403 | 404:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message=_get_error_message(response),
raw_response=response,
)
case 200 | 201:
message = "Record updated successfully" if response.status_code == 200 else "Record created successfully"
return ApiResponse(
status_code=response.status_code,
data=AddOrUpdateRecordResponse.model_validate(response.json()),
message=message,
raw_response=response,
)
case _:
return ApiResponse(status_code=response.status_code, is_successful=False, raw_response=response)
def handle_delete_records_by_ids_response(response: httpx.Response) -> ApiResponse[None]:
match response.status_code:
case 400:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Invalid request provided",
raw_response=response,
)
case 401:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Unauthorized request",
raw_response=response,
)
case 403:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message=_get_error_message(response),
raw_response=response,
)
case 404:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Records could not be found",
raw_response=response,
)
case 204:
return ApiResponse(
status_code=response.status_code, message="Record(s) deleted successfully", raw_response=response
)
case _:
return ApiResponse(status_code=response.status_code, is_successful=False, raw_response=response)
def handle_get_report_by_id_response(response: httpx.Response) -> ApiResponse[GetReportByIdResponse]:
match response.status_code:
case 400:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Invalid request based on underlying data",
raw_response=response,
)
case 401:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Unauthorized request",
raw_response=response,
)
case 403:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message=_get_error_message(response),
raw_response=response,
)
case 404:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Report could not be found",
raw_response=response,
)
case 200:
return ApiResponse(
status_code=response.status_code,
data=GetReportByIdResponse.model_validate(response.json()),
raw_response=response,
)
case _:
return ApiResponse(status_code=response.status_code, is_successful=False, raw_response=response)
def handle_get_reports_by_app_id_response(response: httpx.Response) -> ApiResponse[GetReportsByAppIdResponse]:
match response.status_code:
case 400:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Client does not have read access to the app.",
raw_response=response,
)
case 401:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message="Unauthorized request",
raw_response=response,
)
case 403:
return ApiResponse(
status_code=response.status_code,
is_successful=False,
message=_get_error_message(response),
raw_response=response,
)
case 200:
return ApiResponse(
status_code=response.status_code,
data=GetReportsByAppIdResponse.model_validate(response.json()),
raw_response=response,
)
case _:
return ApiResponse(status_code=response.status_code, is_successful=False, raw_response=response)
src/onspring_api_sdk/async_client.py
+368
View File
@@ -0,0 +1,368 @@
"""Async HTTP client for the Onspring API v2."""
import asyncio
import json
from collections.abc import Mapping
from types import MappingProxyType
from typing import Final
import httpx
from onspring_api_sdk._responses import (
handle_add_or_update_list_item_response,
handle_add_or_update_record_response,
handle_delete_file_by_id_response,
handle_delete_list_item_response,
handle_delete_record_by_id_response,
handle_delete_records_by_ids_response,
handle_get_app_by_id_response,
handle_get_apps_by_ids_response,
handle_get_apps_response,
handle_get_field_by_id_response,
handle_get_fields_by_app_id_response,
handle_get_fields_by_ids_response,
handle_get_file_by_id_response,
handle_get_file_info_by_id_response,
handle_get_record_by_id_response,
handle_get_records_by_app_id_response,
handle_get_records_by_ids_response,
handle_get_report_by_id_response,
handle_get_reports_by_app_id_response,
handle_query_records_response,
handle_save_file_response,
)
from onspring_api_sdk.endpoints import (
add_or_update_list_item_endpoint,
add_or_update_record_endpoint,
delete_file_by_id_endpoint,
delete_list_item_endpoint,
delete_record_by_id_endpoint,
delete_records_by_ids_endpoint,
get_app_by_id_endpoint,
get_apps_by_ids_endpoint,
get_apps_endpoint,
get_field_by_id_endpoint,
get_fields_by_app_id_endpoint,
get_fields_by_ids_endpoint,
get_file_by_id_endpoint,
get_file_info_by_id_endpoint,
get_ping_endpoint,
get_record_by_id_endpoint,
get_records_by_app_id_endpoint,
get_records_by_ids_endpoint,
get_report_by_id_endpoint,
get_reports_by_app_id_endpoint,
query_records_endpoint,
save_file_endpoint,
)
from onspring_api_sdk.models import (
AddOrUpdateListItemResponse,
AddOrUpdateRecordResponse,
ApiResponse,
DeleteBatchRecordsRequest,
GetAppByIdResponse,
GetAppsByIdsResponse,
GetAppsResponse,
GetBatchRecordsRequest,
GetBatchRecordsResponse,
GetFieldByIdResponse,
GetFieldsByAppIdResponse,
GetFieldsByIdsResponse,
GetFileByIdResponse,
GetFileInfoByIdResponse,
GetRecordByIdRequest,
GetRecordsByAppRequest,
GetRecordsResponse,
GetReportByIdRequest,
GetReportByIdResponse,
GetReportsByAppIdResponse,
ListItemRequest,
PagingRequest,
QueryRecordsRequest,
Record,
SaveFileRequest,
SaveFileResponse,
)
API_VERSION = "2"
CONTENT_TYPE_JSON = "application/json"
_JSON_HEADERS: Final[Mapping[str, str]] = MappingProxyType({"Content-Type": CONTENT_TYPE_JSON})
class AsyncOnspringClient:
"""Async client for interacting with the Onspring API v2."""
def __init__(self, url: str, key: str):
"""Initialize the client with a base URL and API key."""
self.client = httpx.AsyncClient(
headers={
"x-apikey": key,
"x-api-version": API_VERSION,
}
)
self.base_url = url
async def aclose(self) -> None:
"""Close the underlying HTTP client."""
await self.client.aclose()
async def __aenter__(self) -> "AsyncOnspringClient":
"""Enter the async runtime context for the client."""
return self
async def __aexit__(self, *args) -> None:
"""Exit the async runtime context and close the client."""
await self.aclose()
async def can_connect(self) -> bool:
"""Ping the API to check connectivity."""
response = await self.client.get(get_ping_endpoint(self.base_url))
return response.status_code == 200
async def get_apps(self, paging_request: PagingRequest | None = None) -> ApiResponse[GetAppsResponse]:
"""Retrieve all apps the API key has access to."""
if paging_request is None:
paging_request = PagingRequest()
response = await self.client.get(
get_apps_endpoint(self.base_url),
params=paging_request.model_dump(by_alias=True, exclude_none=True),
)
return handle_get_apps_response(response)
async def get_app_by_id(self, app_id: int) -> ApiResponse[GetAppByIdResponse]:
"""Retrieve an app by its ID."""
response = await self.client.get(get_app_by_id_endpoint(self.base_url, app_id))
return handle_get_app_by_id_response(response)
async def get_apps_by_ids(self, app_ids: list[int]) -> ApiResponse[GetAppsByIdsResponse]:
"""Retrieve multiple apps by their IDs."""
if not isinstance(app_ids, (list, tuple)):
return ApiResponse(status_code=400, is_successful=False, message="App ids should be of type list or tuple")
response = await self.client.post(
get_apps_by_ids_endpoint(self.base_url),
content=json.dumps(app_ids),
headers=_JSON_HEADERS,
)
return handle_get_apps_by_ids_response(response)
async def get_field_by_id(self, field_id: int) -> ApiResponse[GetFieldByIdResponse]:
"""Retrieve a field by its ID."""
response = await self.client.get(get_field_by_id_endpoint(self.base_url, field_id))
return handle_get_field_by_id_response(response)
async def get_fields_by_ids(self, field_ids: list[int]) -> ApiResponse[GetFieldsByIdsResponse]:
"""Retrieve multiple fields by their IDs."""
if not isinstance(field_ids, (list, tuple)):
return ApiResponse(
status_code=400, is_successful=False, message="Field ids should be of type list or tuple"
)
response = await self.client.post(
get_fields_by_ids_endpoint(self.base_url),
content=json.dumps(field_ids),
headers=_JSON_HEADERS,
)
return handle_get_fields_by_ids_response(response)
async def get_fields_by_app_id(
self, app_id: int, paging_request: PagingRequest | None = None
) -> ApiResponse[GetFieldsByAppIdResponse]:
"""Retrieve all fields for a given app."""
if paging_request is None:
paging_request = PagingRequest()
response = await self.client.get(
get_fields_by_app_id_endpoint(self.base_url, app_id),
params=paging_request.model_dump(by_alias=True, exclude_none=True),
)
return handle_get_fields_by_app_id_response(response)
async def get_file_info_by_id(
self, record_id: int, field_id: int, file_id: int
) -> ApiResponse[GetFileInfoByIdResponse]:
"""Retrieve file metadata for a file attached to a record."""
response = await self.client.get(get_file_info_by_id_endpoint(self.base_url, record_id, field_id, file_id))
return handle_get_file_info_by_id_response(response)
async def delete_file_by_id(self, record_id: int, field_id: int, file_id: int) -> ApiResponse[None]:
"""Delete a file attached to a record."""
response = await self.client.delete(delete_file_by_id_endpoint(self.base_url, record_id, field_id, file_id))
return handle_delete_file_by_id_response(response)
async def get_file_by_id(self, record_id: int, field_id: int, file_id: int) -> ApiResponse[GetFileByIdResponse]:
"""Download a file attached to a record."""
response = await self.client.get(get_file_by_id_endpoint(self.base_url, record_id, field_id, file_id))
return handle_get_file_by_id_response(response)
async def save_file(self, save_file_request: SaveFileRequest) -> ApiResponse[SaveFileResponse]:
"""Upload a file to a record."""
endpoint = save_file_endpoint(self.base_url)
def _read_file() -> bytes:
with open(save_file_request.file_path, "rb") as f:
return f.read()
file_content = await asyncio.to_thread(_read_file)
files = {
"File": (
save_file_request.file_name,
file_content,
save_file_request.content_type,
),
}
data = save_file_request.model_dump(
by_alias=True, exclude={"file_name", "file_path", "content_type"}, exclude_none=True
)
response = await self.client.post(endpoint, data=data, files=files)
return handle_save_file_response(response)
async def add_or_update_list_item(
self, list_item_request: ListItemRequest
) -> ApiResponse[AddOrUpdateListItemResponse]:
"""Add or update a list item value."""
endpoint = add_or_update_list_item_endpoint(self.base_url, list_item_request.list_id)
payload = list_item_request.model_dump(by_alias=True, exclude={"list_id"}, exclude_none=True, mode="json")
response = await self.client.put(
endpoint,
content=json.dumps(payload),
headers=_JSON_HEADERS,
)
return handle_add_or_update_list_item_response(response)
async def delete_list_item(self, list_id: int, item_id: str) -> ApiResponse[None]:
"""Delete a list item by its ID."""
response = await self.client.delete(delete_list_item_endpoint(self.base_url, list_id, item_id))
return handle_delete_list_item_response(response)
async def get_records_by_app_id(self, request: GetRecordsByAppRequest) -> ApiResponse[GetRecordsResponse]:
"""Retrieve records from an app with optional filtering and paging."""
params = request.model_dump(by_alias=True, exclude={"app_id"}, exclude_none=True)
field_ids = params.pop("fieldIds", None)
if field_ids:
params["fieldIds"] = ",".join(str(i) for i in field_ids)
response = await self.client.get(
get_records_by_app_id_endpoint(self.base_url, request.app_id),
params=params,
)
return handle_get_records_by_app_id_response(response)
async def get_record_by_id(self, request: GetRecordByIdRequest) -> ApiResponse[Record]:
"""Retrieve a single record by its ID."""
params = request.model_dump(by_alias=True, exclude={"app_id", "record_id"}, exclude_none=True)
field_ids = params.pop("fieldIds", None)
if field_ids:
params["fieldIds"] = ",".join(str(i) for i in field_ids)
response = await self.client.get(
get_record_by_id_endpoint(self.base_url, request.app_id, request.record_id),
params=params,
)
return handle_get_record_by_id_response(response)
async def delete_record_by_id(self, app_id: int, record_id: int) -> ApiResponse[None]:
"""Delete a single record by its ID."""
response = await self.client.delete(delete_record_by_id_endpoint(self.base_url, app_id, record_id))
return handle_delete_record_by_id_response(response)
async def get_records_by_ids(self, request: GetBatchRecordsRequest) -> ApiResponse[GetBatchRecordsResponse]:
"""Retrieve multiple records by their IDs."""
response = await self.client.post(
get_records_by_ids_endpoint(self.base_url),
content=json.dumps(request.model_dump(by_alias=True, exclude_none=True, mode="json")),
headers=_JSON_HEADERS,
)
return handle_get_records_by_ids_response(response)
async def query_records(self, request: QueryRecordsRequest) -> ApiResponse[GetRecordsResponse]:
"""Query records using a structured query."""
exclude = {"page_number", "page_size"}
payload = request.model_dump(by_alias=True, exclude=exclude, exclude_none=True, mode="json")
params = {"pageNumber": request.page_number, "pageSize": request.page_size}
response = await self.client.post(
query_records_endpoint(self.base_url),
content=json.dumps(payload),
params=params,
headers=_JSON_HEADERS,
)
return handle_query_records_response(response)
async def add_or_update_record(self, record: Record) -> ApiResponse[AddOrUpdateRecordResponse]:
"""Add or update a record."""
fields_dict = {}
for field in record.fields:
fields_dict[field.field_id] = field.value
payload = record.model_dump(by_alias=True, exclude={"fields"}, exclude_none=True, mode="json")
payload["fields"] = fields_dict
response = await self.client.put(
add_or_update_record_endpoint(self.base_url),
content=json.dumps(payload, default=str),
headers=_JSON_HEADERS,
)
return handle_add_or_update_record_response(response)
async def delete_records_by_ids(self, request: DeleteBatchRecordsRequest) -> ApiResponse[None]:
"""Delete multiple records by their IDs."""
response = await self.client.post(
delete_records_by_ids_endpoint(self.base_url),
content=json.dumps(request.model_dump(by_alias=True, exclude_none=True, mode="json")),
headers=_JSON_HEADERS,
)
return handle_delete_records_by_ids_response(response)
async def get_report_by_id(self, request: GetReportByIdRequest) -> ApiResponse[GetReportByIdResponse]:
"""Retrieve a report by its ID."""
params = request.model_dump(by_alias=True, exclude={"report_id"}, exclude_none=True)
response = await self.client.get(
get_report_by_id_endpoint(self.base_url, request.report_id),
params=params,
)
return handle_get_report_by_id_response(response)
async def get_reports_by_app_id(
self, app_id: int, paging_request: PagingRequest | None = None
) -> ApiResponse[GetReportsByAppIdResponse]:
"""Retrieve all reports for a given app."""
if paging_request is None:
paging_request = PagingRequest()
response = await self.client.get(
get_reports_by_app_id_endpoint(self.base_url, app_id),
params=paging_request.model_dump(by_alias=True, exclude_none=True),
)
return handle_get_reports_by_app_id_response(response)
src/onspring_api_sdk/client.py
+359
View File
@@ -0,0 +1,359 @@
"""Sync HTTP client for the Onspring API v2."""
import json
from collections.abc import Mapping
from types import MappingProxyType
from typing import Final
import httpx
from onspring_api_sdk._responses import (
handle_add_or_update_list_item_response,
handle_add_or_update_record_response,
handle_delete_file_by_id_response,
handle_delete_list_item_response,
handle_delete_record_by_id_response,
handle_delete_records_by_ids_response,
handle_get_app_by_id_response,
handle_get_apps_by_ids_response,
handle_get_apps_response,
handle_get_field_by_id_response,
handle_get_fields_by_app_id_response,
handle_get_fields_by_ids_response,
handle_get_file_by_id_response,
handle_get_file_info_by_id_response,
handle_get_record_by_id_response,
handle_get_records_by_app_id_response,
handle_get_records_by_ids_response,
handle_get_report_by_id_response,
handle_get_reports_by_app_id_response,
handle_query_records_response,
handle_save_file_response,
)
from onspring_api_sdk.endpoints import (
add_or_update_list_item_endpoint,
add_or_update_record_endpoint,
delete_file_by_id_endpoint,
delete_list_item_endpoint,
delete_record_by_id_endpoint,
delete_records_by_ids_endpoint,
get_app_by_id_endpoint,
get_apps_by_ids_endpoint,
get_apps_endpoint,
get_field_by_id_endpoint,
get_fields_by_app_id_endpoint,
get_fields_by_ids_endpoint,
get_file_by_id_endpoint,
get_file_info_by_id_endpoint,
get_ping_endpoint,
get_record_by_id_endpoint,
get_records_by_app_id_endpoint,
get_records_by_ids_endpoint,
get_report_by_id_endpoint,
get_reports_by_app_id_endpoint,
query_records_endpoint,
save_file_endpoint,
)
from onspring_api_sdk.models import (
AddOrUpdateListItemResponse,
AddOrUpdateRecordResponse,
ApiResponse,
DeleteBatchRecordsRequest,
GetAppByIdResponse,
GetAppsByIdsResponse,
GetAppsResponse,
GetBatchRecordsRequest,
GetBatchRecordsResponse,
GetFieldByIdResponse,
GetFieldsByAppIdResponse,
GetFieldsByIdsResponse,
GetFileByIdResponse,
GetFileInfoByIdResponse,
GetRecordByIdRequest,
GetRecordsByAppRequest,
GetRecordsResponse,
GetReportByIdRequest,
GetReportByIdResponse,
GetReportsByAppIdResponse,
ListItemRequest,
PagingRequest,
QueryRecordsRequest,
Record,
SaveFileRequest,
SaveFileResponse,
)
API_VERSION = "2"
CONTENT_TYPE_JSON = "application/json"
_JSON_HEADERS: Final[Mapping[str, str]] = MappingProxyType({"Content-Type": CONTENT_TYPE_JSON})
class OnspringClient:
"""Sync client for interacting with the Onspring API v2."""
def __init__(self, url: str, key: str):
"""Initialize the client with a base URL and API key."""
self.client = httpx.Client(
headers={
"x-apikey": key,
"x-api-version": API_VERSION,
}
)
self.base_url = url
def close(self) -> None:
"""Close the underlying HTTP client."""
self.client.close()
def __enter__(self) -> "OnspringClient":
"""Enter the runtime context for the client."""
return self
def __exit__(self, *args) -> None:
"""Exit the runtime context and close the client."""
self.close()
def can_connect(self) -> bool:
"""Ping the API to check connectivity."""
response = self.client.get(get_ping_endpoint(self.base_url))
return response.status_code == 200
def get_apps(self, paging_request: PagingRequest | None = None) -> ApiResponse[GetAppsResponse]:
"""Get all apps with optional paging."""
if paging_request is None:
paging_request = PagingRequest()
response = self.client.get(
get_apps_endpoint(self.base_url),
params=paging_request.model_dump(by_alias=True, exclude_none=True),
)
return handle_get_apps_response(response)
def get_app_by_id(self, app_id: int) -> ApiResponse[GetAppByIdResponse]:
"""Get an app by its ID."""
response = self.client.get(get_app_by_id_endpoint(self.base_url, app_id))
return handle_get_app_by_id_response(response)
def get_apps_by_ids(self, app_ids: list[int]) -> ApiResponse[GetAppsByIdsResponse]:
"""Get multiple apps by their IDs."""
if not isinstance(app_ids, (list, tuple)):
return ApiResponse(status_code=400, is_successful=False, message="App ids should be of type list or tuple")
response = self.client.post(
get_apps_by_ids_endpoint(self.base_url),
content=json.dumps(app_ids),
headers=_JSON_HEADERS,
)
return handle_get_apps_by_ids_response(response)
def get_field_by_id(self, field_id: int) -> ApiResponse[GetFieldByIdResponse]:
"""Get a field by its ID."""
response = self.client.get(get_field_by_id_endpoint(self.base_url, field_id))
return handle_get_field_by_id_response(response)
def get_fields_by_ids(self, field_ids: list[int]) -> ApiResponse[GetFieldsByIdsResponse]:
"""Get multiple fields by their IDs."""
if not isinstance(field_ids, (list, tuple)):
return ApiResponse(
status_code=400, is_successful=False, message="Field ids should be of type list or tuple"
)
response = self.client.post(
get_fields_by_ids_endpoint(self.base_url),
content=json.dumps(field_ids),
headers=_JSON_HEADERS,
)
return handle_get_fields_by_ids_response(response)
def get_fields_by_app_id(
self, app_id: int, paging_request: PagingRequest | None = None
) -> ApiResponse[GetFieldsByAppIdResponse]:
"""Get all fields for an app with optional paging."""
if paging_request is None:
paging_request = PagingRequest()
response = self.client.get(
get_fields_by_app_id_endpoint(self.base_url, app_id),
params=paging_request.model_dump(by_alias=True, exclude_none=True),
)
return handle_get_fields_by_app_id_response(response)
def get_file_info_by_id(self, record_id: int, field_id: int, file_id: int) -> ApiResponse[GetFileInfoByIdResponse]:
"""Get file metadata by record, field, and file IDs."""
response = self.client.get(get_file_info_by_id_endpoint(self.base_url, record_id, field_id, file_id))
return handle_get_file_info_by_id_response(response)
def delete_file_by_id(self, record_id: int, field_id: int, file_id: int) -> ApiResponse[None]:
"""Delete a file by record, field, and file IDs."""
response = self.client.delete(delete_file_by_id_endpoint(self.base_url, record_id, field_id, file_id))
return handle_delete_file_by_id_response(response)
def get_file_by_id(self, record_id: int, field_id: int, file_id: int) -> ApiResponse[GetFileByIdResponse]:
"""Get a file by record, field, and file IDs."""
response = self.client.get(get_file_by_id_endpoint(self.base_url, record_id, field_id, file_id))
return handle_get_file_by_id_response(response)
def save_file(self, save_file_request: SaveFileRequest) -> ApiResponse[SaveFileResponse]:
"""Save a file to a record."""
endpoint = save_file_endpoint(self.base_url)
with open(save_file_request.file_path, "rb") as f:
file_content = f.read()
files = {
"File": (
save_file_request.file_name,
file_content,
save_file_request.content_type,
),
}
data = save_file_request.model_dump(
by_alias=True, exclude={"file_name", "file_path", "content_type"}, exclude_none=True
)
response = self.client.post(endpoint, data=data, files=files)
return handle_save_file_response(response)
def add_or_update_list_item(self, list_item_request: ListItemRequest) -> ApiResponse[AddOrUpdateListItemResponse]:
"""Add or update a list item value."""
endpoint = add_or_update_list_item_endpoint(self.base_url, list_item_request.list_id)
payload = list_item_request.model_dump(by_alias=True, exclude={"list_id"}, exclude_none=True, mode="json")
response = self.client.put(
endpoint,
content=json.dumps(payload),
headers=_JSON_HEADERS,
)
return handle_add_or_update_list_item_response(response)
def delete_list_item(self, list_id: int, item_id: str) -> ApiResponse[None]:
"""Delete a list item by list and item IDs."""
response = self.client.delete(delete_list_item_endpoint(self.base_url, list_id, item_id))
return handle_delete_list_item_response(response)
def get_records_by_app_id(self, request: GetRecordsByAppRequest) -> ApiResponse[GetRecordsResponse]:
"""Get records for an app with optional filtering and paging."""
params = request.model_dump(by_alias=True, exclude={"app_id"}, exclude_none=True)
field_ids = params.pop("fieldIds", None)
if field_ids:
params["fieldIds"] = ",".join(str(i) for i in field_ids)
response = self.client.get(
get_records_by_app_id_endpoint(self.base_url, request.app_id),
params=params,
)
return handle_get_records_by_app_id_response(response)
def get_record_by_id(self, request: GetRecordByIdRequest) -> ApiResponse[Record]:
"""Get a record by its app and record IDs."""
params = request.model_dump(by_alias=True, exclude={"app_id", "record_id"}, exclude_none=True)
field_ids = params.pop("fieldIds", None)
if field_ids:
params["fieldIds"] = ",".join(str(i) for i in field_ids)
response = self.client.get(
get_record_by_id_endpoint(self.base_url, request.app_id, request.record_id),
params=params,
)
return handle_get_record_by_id_response(response)
def delete_record_by_id(self, app_id: int, record_id: int) -> ApiResponse[None]:
"""Delete a record by its app and record IDs."""
response = self.client.delete(delete_record_by_id_endpoint(self.base_url, app_id, record_id))
return handle_delete_record_by_id_response(response)
def get_records_by_ids(self, request: GetBatchRecordsRequest) -> ApiResponse[GetBatchRecordsResponse]:
"""Get multiple records by their IDs."""
response = self.client.post(
get_records_by_ids_endpoint(self.base_url),
content=json.dumps(request.model_dump(by_alias=True, exclude_none=True, mode="json")),
headers=_JSON_HEADERS,
)
return handle_get_records_by_ids_response(response)
def query_records(self, request: QueryRecordsRequest) -> ApiResponse[GetRecordsResponse]:
"""Query records using a structured query request."""
exclude = {"page_number", "page_size"}
payload = request.model_dump(by_alias=True, exclude=exclude, exclude_none=True, mode="json")
params = {"pageNumber": request.page_number, "pageSize": request.page_size}
response = self.client.post(
query_records_endpoint(self.base_url),
content=json.dumps(payload),
params=params,
headers=_JSON_HEADERS,
)
return handle_query_records_response(response)
def add_or_update_record(self, record: Record) -> ApiResponse[AddOrUpdateRecordResponse]:
"""Add or update a record."""
fields_dict = {}
for field in record.fields:
fields_dict[field.field_id] = field.value
payload = record.model_dump(by_alias=True, exclude={"fields"}, exclude_none=True, mode="json")
payload["fields"] = fields_dict
response = self.client.put(
add_or_update_record_endpoint(self.base_url),
content=json.dumps(payload, default=str),
headers=_JSON_HEADERS,
)
return handle_add_or_update_record_response(response)
def delete_records_by_ids(self, request: DeleteBatchRecordsRequest) -> ApiResponse[None]:
"""Delete multiple records by their IDs."""
response = self.client.post(
delete_records_by_ids_endpoint(self.base_url),
content=json.dumps(request.model_dump(by_alias=True, exclude_none=True, mode="json")),
headers=_JSON_HEADERS,
)
return handle_delete_records_by_ids_response(response)
def get_report_by_id(self, request: GetReportByIdRequest) -> ApiResponse[GetReportByIdResponse]:
"""Get a report by its ID."""
params = request.model_dump(by_alias=True, exclude={"report_id"}, exclude_none=True)
response = self.client.get(
get_report_by_id_endpoint(self.base_url, request.report_id),
params=params,
)
return handle_get_report_by_id_response(response)
def get_reports_by_app_id(
self, app_id: int, paging_request: PagingRequest | None = None
) -> ApiResponse[GetReportsByAppIdResponse]:
"""Get all reports for an app with optional paging."""
if paging_request is None:
paging_request = PagingRequest()
response = self.client.get(
get_reports_by_app_id_endpoint(self.base_url, app_id),
params=paging_request.model_dump(by_alias=True, exclude_none=True),
)
return handle_get_reports_by_app_id_response(response)
src/onspring_api_sdk/endpoints.py
+111
View File
@@ -0,0 +1,111 @@
"""URL builder functions for all Onspring API v2 endpoints."""
def get_ping_endpoint(base_url: str) -> str:
"""Build the ping endpoint URL."""
return f"{base_url}/Ping"
def get_apps_endpoint(base_url: str) -> str:
"""Build the get-apps endpoint URL."""
return f"{base_url}/Apps"
def get_app_by_id_endpoint(base_url: str, app_id: int) -> str:
"""Build the get-app-by-id endpoint URL."""
return f"{base_url}/Apps/id/{app_id}"
def get_apps_by_ids_endpoint(base_url: str) -> str:
"""Build the get-apps-by-ids endpoint URL."""
return f"{base_url}/Apps/batch-get"
def get_field_by_id_endpoint(base_url: str, field_id: int) -> str:
"""Build the get-field-by-id endpoint URL."""
return f"{base_url}/Fields/id/{field_id}"
def get_fields_by_ids_endpoint(base_url: str) -> str:
"""Build the get-fields-by-ids endpoint URL."""
return f"{base_url}/Fields/batch-get"
def get_fields_by_app_id_endpoint(base_url: str, app_id: int) -> str:
"""Build the get-fields-by-app-id endpoint URL."""
return f"{base_url}/Fields/appId/{app_id}"
def get_file_info_by_id_endpoint(base_url: str, record_id: int, field_id: int, file_id: int) -> str:
"""Build the get-file-info-by-id endpoint URL."""
return f"{base_url}/Files/recordId/{record_id}/fieldId/{field_id}/fileId/{file_id}"
def delete_file_by_id_endpoint(base_url: str, record_id: int, field_id: int, file_id: int) -> str:
"""Build the delete-file-by-id endpoint URL."""
return f"{base_url}/Files/recordId/{record_id}/fieldId/{field_id}/fileId/{file_id}"
def get_file_by_id_endpoint(base_url: str, record_id: int, field_id: int, file_id: int) -> str:
"""Build the get-file-by-id endpoint URL."""
return f"{base_url}/Files/recordId/{record_id}/fieldId/{field_id}/fileId/{file_id}/file"
def save_file_endpoint(base_url: str) -> str:
"""Build the save-file endpoint URL."""
return f"{base_url}/Files"
def add_or_update_list_item_endpoint(base_url: str, list_id: int) -> str:
"""Build the add-or-update-list-item endpoint URL."""
return f"{base_url}/Lists/id/{list_id}/items"
def delete_list_item_endpoint(base_url: str, list_id: int, item_id: str) -> str:
"""Build the delete-list-item endpoint URL."""
return f"{base_url}/Lists/id/{list_id}/itemId/{item_id}"
def get_records_by_app_id_endpoint(base_url: str, app_id: int) -> str:
"""Build the get-records-by-app-id endpoint URL."""
return f"{base_url}/Records/appId/{app_id}"
def get_record_by_id_endpoint(base_url: str, app_id: int, record_id: int) -> str:
"""Build the get-record-by-id endpoint URL."""
return f"{base_url}/Records/appId/{app_id}/recordId/{record_id}"
def delete_record_by_id_endpoint(base_url: str, app_id: int, record_id: int) -> str:
"""Build the delete-record-by-id endpoint URL."""
return f"{base_url}/Records/appId/{app_id}/recordId/{record_id}"
def get_records_by_ids_endpoint(base_url: str) -> str:
"""Build the get-records-by-ids endpoint URL."""
return f"{base_url}/Records/batch-get"
def query_records_endpoint(base_url: str) -> str:
"""Build the query-records endpoint URL."""
return f"{base_url}/Records/Query"
def add_or_update_record_endpoint(base_url: str) -> str:
"""Build the add-or-update-record endpoint URL."""
return f"{base_url}/Records"
def delete_records_by_ids_endpoint(base_url: str) -> str:
"""Build the delete-records-by-ids endpoint URL."""
return f"{base_url}/Records/batch-delete"
def get_report_by_id_endpoint(base_url: str, report_id: int) -> str:
"""Build the get-report-by-id endpoint URL."""
return f"{base_url}/Reports/id/{report_id}"
def get_reports_by_app_id_endpoint(base_url: str, app_id: int) -> str:
"""Build the get-reports-by-app-id endpoint URL."""
return f"{base_url}/Reports/appId/{app_id}"
src/onspring_api_sdk/enums.py
+54
View File
@@ -0,0 +1,54 @@
"""Enumerations for Onspring API data types and timespan configuration."""
from enum import Enum
class DataFormat(Enum):
"""The possible data format types for record field values."""
Raw: int = 0
Formatted: int = 1
class ReportDataType(Enum):
"""The possible report data types for reports."""
ReportData: int = 0
ChartData: int = 1
class ResultValueType(Enum):
"""The possible types for record field values."""
String: int = 0
Integer: int = 1
Decimal: int = 2
Date: int = 3
TimeSpan: int = 4
Guid: int = 5
StringList: int = 6
IntegerList: int = 7
GuidList: int = 8
AttachmentList: int = 9
ScoringGroupList: int = 10
FileList: int = 11
class Increment(Enum):
"""Possible increment values for timespan data in an Onspring timespan field."""
Seconds: str = "Second(s)"
Minutes: str = "Minute(s)"
Hours: str = "Hour(s)"
Days: str = "Day(s)"
Weeks: str = "Week(s)"
Months: str = "Month(s)"
Years: str = "Year(s)"
class Recurrence(Enum):
"""Possible recurrence values for timespan data in an Onspring timespan field."""
Empty: str = "None"
EndByDate: str = "EndByDate"
EndAfterOccurrences: str = "EndAfterOccurrences"
src/onspring_api_sdk/errors.py
+32
View File
@@ -0,0 +1,32 @@
"""Custom exception hierarchy for Onspring API errors."""
import json
import httpx
def _get_error_message(response: httpx.Response) -> str | None:
"""Safely extract an error message from an API response body."""
try:
body = response.json()
if isinstance(body, dict):
return body.get("message")
except (json.JSONDecodeError, ValueError, AttributeError):
pass
return None
class OnspringError(Exception):
"""Base exception for all Onspring API errors."""
class OnspringAuthenticationError(OnspringError):
"""Raised when the API returns a 401 or 403 status code."""
class OnspringNotFoundError(OnspringError):
"""Raised when the API returns a 404 status code."""
class OnspringRateLimitError(OnspringError):
"""Raised when the API returns a 429 status code."""
src/onspring_api_sdk/models/__init__.py
+106
View File
@@ -0,0 +1,106 @@
"""Re-exports for all Pydantic models used by the Onspring API SDK."""
from onspring_api_sdk.models.app import App, GetAppByIdResponse, GetAppsByIdsResponse, GetAppsResponse
from onspring_api_sdk.models.common import ApiResponse, PagingRequest
from onspring_api_sdk.models.field import (
GetFieldByIdResponse,
GetFieldsByAppIdResponse,
GetFieldsByIdsResponse,
ListValue,
OnspringField,
)
from onspring_api_sdk.models.file import (
File,
FileInfo,
GetFileByIdResponse,
GetFileInfoByIdResponse,
SaveFileRequest,
SaveFileResponse,
)
from onspring_api_sdk.models.list import AddOrUpdateListItemResponse, ListItemRequest
from onspring_api_sdk.models.record import (
AddOrUpdateRecordResponse,
Attachment,
AttachmentListValue,
DateFieldValue,
DecimalFieldValue,
DeleteBatchRecordsRequest,
FileListValue,
GetBatchRecordsRequest,
GetBatchRecordsResponse,
GetRecordByIdRequest,
GetRecordsByAppRequest,
GetRecordsResponse,
GuidFieldValue,
GuidListValue,
IntegerFieldValue,
IntegerListValue,
QueryRecordsRequest,
Record,
RecordFieldValue,
ScoringGroup,
ScoringGroupListValue,
StringFieldValue,
StringListValue,
TimeSpanData,
TimeSpanValue,
)
from onspring_api_sdk.models.report import (
GetReportByIdRequest,
GetReportByIdResponse,
GetReportsByAppIdResponse,
Report,
Row,
)
__all__ = [
"ApiResponse",
"PagingRequest",
"App",
"GetAppsResponse",
"GetAppByIdResponse",
"GetAppsByIdsResponse",
"ListValue",
"OnspringField",
"GetFieldByIdResponse",
"GetFieldsByIdsResponse",
"GetFieldsByAppIdResponse",
"RecordFieldValue",
"StringFieldValue",
"IntegerFieldValue",
"DecimalFieldValue",
"DateFieldValue",
"GuidFieldValue",
"TimeSpanData",
"TimeSpanValue",
"StringListValue",
"IntegerListValue",
"GuidListValue",
"Attachment",
"AttachmentListValue",
"ScoringGroup",
"ScoringGroupListValue",
"FileListValue",
"Record",
"GetRecordsByAppRequest",
"QueryRecordsRequest",
"GetRecordsResponse",
"GetRecordByIdRequest",
"GetBatchRecordsRequest",
"GetBatchRecordsResponse",
"AddOrUpdateRecordResponse",
"DeleteBatchRecordsRequest",
"Row",
"Report",
"GetReportByIdRequest",
"GetReportByIdResponse",
"GetReportsByAppIdResponse",
"File",
"FileInfo",
"GetFileInfoByIdResponse",
"GetFileByIdResponse",
"SaveFileRequest",
"SaveFileResponse",
"ListItemRequest",
"AddOrUpdateListItemResponse",
]
src/onspring_api_sdk/models/app.py
+38
View File
@@ -0,0 +1,38 @@
"""Pydantic models for Onspring app API responses."""
from pydantic import BaseModel, ConfigDict, Field
class App(BaseModel):
"""Represents an Onspring app."""
href: str
id: int
name: str
class GetAppsResponse(BaseModel):
"""Paginated response containing a list of apps."""
model_config = ConfigDict(populate_by_name=True)
page_number: int = Field(alias="pageNumber")
page_size: int = Field(alias="pageSize")
total_pages: int = Field(alias="totalPages")
total_records: int = Field(alias="totalRecords")
apps: list[App] = Field(alias="items")
class GetAppByIdResponse(BaseModel):
"""Response containing a single app."""
app: App
class GetAppsByIdsResponse(BaseModel):
"""Response containing a list of apps for requested IDs."""
model_config = ConfigDict(populate_by_name=True)
count: int
apps: list[App] = Field(alias="items")
src/onspring_api_sdk/models/common.py
+55
View File
@@ -0,0 +1,55 @@
"""Shared Pydantic models for paging and generic API responses."""
from typing import Generic, Optional, TypeVar
from httpx import Response
from pydantic import BaseModel, ConfigDict, Field, model_validator
from onspring_api_sdk.errors import (
OnspringAuthenticationError,
OnspringError,
OnspringNotFoundError,
OnspringRateLimitError,
)
T = TypeVar("T")
class PagingRequest(BaseModel):
"""Paging parameters for paginated API requests."""
model_config = ConfigDict(populate_by_name=True)
page_number: int = Field(alias="pageNumber", default=1)
page_size: int = Field(alias="pageSize", default=50)
class ApiResponse(BaseModel, Generic[T]):
"""Generic wrapper for all API responses with status and error handling."""
model_config = ConfigDict(arbitrary_types_allowed=True)
status_code: int
is_successful: bool = True
message: Optional[str] = None
data: Optional[T] = None
raw_response: Optional[Response] = None
@model_validator(mode="before")
@classmethod
def set_is_successful(cls, data):
"""Automatically infer is_successful from status_code if not provided."""
if isinstance(data, dict) and "is_successful" not in data and "status_code" in data:
data["is_successful"] = int(data["status_code"]) < 400
return data
def raise_for_status(self):
"""Raise the appropriate exception if the request was not successful."""
if not self.is_successful:
if self.status_code in (401, 403):
raise OnspringAuthenticationError(self.message or "Authentication failed")
if self.status_code == 404:
raise OnspringNotFoundError(self.message or "Resource not found")
if self.status_code == 429:
raise OnspringRateLimitError(self.message or "Rate limit exceeded")
raise OnspringError(self.message or f"Request failed with status {self.status_code}")
src/onspring_api_sdk/models/field.py
+63
View File
@@ -0,0 +1,63 @@
"""Pydantic models for Onspring field API responses."""
import uuid
from typing import Optional
from pydantic import BaseModel, ConfigDict, Field
class ListValue(BaseModel):
"""A selectable value within a list field definition."""
model_config = ConfigDict(populate_by_name=True)
id: uuid.UUID
name: str
sort_order: int = Field(alias="sortOrder")
numeric_value: Optional[int] = Field(default=None, alias="numericValue")
color: Optional[str] = None
class OnspringField(BaseModel):
"""Represents an Onspring field definition."""
model_config = ConfigDict(populate_by_name=True)
id: int
app_id: int = Field(alias="appId")
name: str
type: str
status: str
is_required: bool = Field(alias="isRequired")
is_unique: bool = Field(alias="isUnique")
list_id: Optional[int] = Field(default=None, alias="listId")
values: Optional[list[ListValue]] = None
multiplicity: Optional[str] = None
output_type: Optional[str] = Field(default=None, alias="outputType")
class GetFieldByIdResponse(BaseModel):
"""Response containing a single field."""
field: OnspringField
class GetFieldsByIdsResponse(BaseModel):
"""Response containing fields for requested IDs."""
model_config = ConfigDict(populate_by_name=True)
count: int
fields: list[OnspringField] = Field(alias="items")
class GetFieldsByAppIdResponse(BaseModel):
"""Paginated response containing fields for an app."""
model_config = ConfigDict(populate_by_name=True)
page_number: int = Field(alias="pageNumber")
page_size: int = Field(alias="pageSize")
total_pages: int = Field(alias="totalPages")
total_records: int = Field(alias="totalRecords")
fields: list[OnspringField] = Field(alias="items")
src/onspring_api_sdk/models/file.py
+65
View File
@@ -0,0 +1,65 @@
"""Pydantic models for Onspring file API requests and responses."""
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, ConfigDict, Field
class File(BaseModel):
"""Represents a file with its metadata and binary content."""
model_config = ConfigDict(populate_by_name=True)
name: str
content_type: str = Field(alias="contentType")
content_length: int = Field(alias="contentLength")
content: bytes
class FileInfo(BaseModel):
"""Metadata about a file stored in Onspring."""
model_config = ConfigDict(populate_by_name=True)
type: str
content_type: str = Field(alias="contentType")
name: str
created_date: Optional[datetime] = Field(default=None, alias="createdDate")
modified_date: Optional[datetime] = Field(default=None, alias="modifiedDate")
owner: str
file_href: str = Field(alias="fileHref")
class GetFileInfoByIdResponse(BaseModel):
"""Response containing file metadata."""
model_config = ConfigDict(populate_by_name=True)
file_info: FileInfo = Field(alias="fileInfo")
class GetFileByIdResponse(BaseModel):
"""Response containing a file with binary content."""
file: File
class SaveFileRequest(BaseModel):
"""Request payload for uploading a file to a record field."""
model_config = ConfigDict(populate_by_name=True)
record_id: int = Field(alias="recordId")
field_id: int = Field(alias="fieldId")
file_name: str = Field(alias="fileName")
file_path: str = Field(alias="filePath")
content_type: str = Field(alias="contentType")
notes: Optional[str] = None
modified_date: Optional[datetime] = Field(default=None, alias="modifiedDate")
class SaveFileResponse(BaseModel):
"""Response containing the ID of a saved file."""
id: int
src/onspring_api_sdk/models/list.py
+24
View File
@@ -0,0 +1,24 @@
"""Pydantic models for Onspring list item API requests and responses."""
import uuid
from typing import Optional
from pydantic import BaseModel, ConfigDict, Field
class ListItemRequest(BaseModel):
"""Request payload for adding or updating a list item."""
model_config = ConfigDict(populate_by_name=True)
list_id: int = Field(alias="listId")
name: str
id: Optional[uuid.UUID] = None
numeric_value: Optional[int] = Field(default=None, alias="numericValue")
color: Optional[str] = None
class AddOrUpdateListItemResponse(BaseModel):
"""Response containing the ID of an added or updated list item."""
id: uuid.UUID
src/onspring_api_sdk/models/record.py
+252
View File
@@ -0,0 +1,252 @@
"""Pydantic models for Onspring record data, requests, and responses."""
import uuid
from datetime import datetime
from decimal import Decimal
from typing import Annotated, Literal, Optional, Union
from pydantic import BaseModel, ConfigDict, Field, Tag
from onspring_api_sdk.enums import DataFormat
class RecordFieldValue(BaseModel):
"""Base model for all record field value types."""
model_config = ConfigDict(populate_by_name=True)
field_id: int = Field(alias="fieldId")
value: object = None
type: str
class StringFieldValue(RecordFieldValue):
"""Field value containing a string."""
type: Literal["String"] = "String"
value: Optional[str] = None
class IntegerFieldValue(RecordFieldValue):
"""Field value containing an integer."""
type: Literal["Integer"] = "Integer"
value: Optional[int] = None
class DecimalFieldValue(RecordFieldValue):
"""Field value containing a decimal number."""
type: Literal["Decimal"] = "Decimal"
value: Optional[Decimal] = None
class DateFieldValue(RecordFieldValue):
"""Field value containing a date."""
type: Literal["Date"] = "Date"
value: Optional[datetime] = None
class GuidFieldValue(RecordFieldValue):
"""Field value containing a GUID."""
type: Literal["Guid"] = "Guid"
value: Optional[uuid.UUID] = None
class TimeSpanData(BaseModel):
"""Time span configuration with increment, recurrence, and end conditions."""
model_config = ConfigDict(populate_by_name=True)
quantity: Optional[Decimal] = None
increment: Optional[str] = None
recurrence: Optional[str] = None
end_by_date: Optional[datetime] = Field(default=None, alias="endByDate")
end_after_occurrences: Optional[int] = Field(default=None, alias="endAfterOccurrences")
class TimeSpanValue(RecordFieldValue):
"""Field value containing a time span."""
type: Literal["TimeSpan"] = "TimeSpan"
value: Optional[TimeSpanData] = None
class StringListValue(RecordFieldValue):
"""Field value containing a list of strings."""
type: Literal["StringList"] = "StringList"
value: Optional[list[str]] = None
class IntegerListValue(RecordFieldValue):
"""Field value containing a list of integers."""
type: Literal["IntegerList"] = "IntegerList"
value: Optional[list[int]] = None
class GuidListValue(RecordFieldValue):
"""Field value containing a list of GUIDs."""
type: Literal["GuidList"] = "GuidList"
value: Optional[list[uuid.UUID]] = None
class Attachment(BaseModel):
"""An attachment associated with a record field."""
model_config = ConfigDict(populate_by_name=True)
file_id: int = Field(alias="fileId")
file_name: str = Field(alias="fileName")
notes: Optional[str] = None
storage_location: str = Field(alias="storageLocation")
class AttachmentListValue(RecordFieldValue):
"""Field value containing a list of attachments."""
type: Literal["AttachmentList"] = "AttachmentList"
value: Optional[list[Attachment]] = None
class ScoringGroup(BaseModel):
"""A scoring group with a list value reference, name, and scores."""
model_config = ConfigDict(populate_by_name=True)
list_value_id: Optional[uuid.UUID] = Field(default=None, alias="listValueId")
name: Optional[str] = None
score: Optional[Decimal] = None
maximum_score: Optional[Decimal] = Field(default=None, alias="maximumScore")
delegate_type: Optional[str] = Field(default=None, alias="delegateType")
class ScoringGroupListValue(RecordFieldValue):
"""Field value containing a list of scoring groups."""
type: Literal["ScoringGroupList"] = "ScoringGroupList"
value: Optional[list[ScoringGroup]] = None
class FileListValue(RecordFieldValue):
"""Field value containing a list of file IDs."""
type: Literal["FileList"] = "FileList"
value: Optional[list[int]] = None
FieldValue = Annotated[
Union[
Annotated[StringFieldValue, Tag("String")],
Annotated[IntegerFieldValue, Tag("Integer")],
Annotated[DecimalFieldValue, Tag("Decimal")],
Annotated[DateFieldValue, Tag("Date")],
Annotated[GuidFieldValue, Tag("Guid")],
Annotated[TimeSpanValue, Tag("TimeSpan")],
Annotated[StringListValue, Tag("StringList")],
Annotated[IntegerListValue, Tag("IntegerList")],
Annotated[GuidListValue, Tag("GuidList")],
Annotated[AttachmentListValue, Tag("AttachmentList")],
Annotated[ScoringGroupListValue, Tag("ScoringGroupList")],
Annotated[FileListValue, Tag("FileList")],
],
Field(discriminator="type"),
]
class Record(BaseModel):
"""A record containing typed field values for a specific app."""
model_config = ConfigDict(populate_by_name=True)
app_id: int = Field(alias="appId")
record_id: Optional[int] = Field(default=None, alias="recordId")
fields: list[FieldValue] = Field(default_factory=list, alias="fieldData")
class GetRecordsByAppRequest(BaseModel):
"""Request parameters for fetching records by app ID."""
model_config = ConfigDict(populate_by_name=True)
app_id: int = Field(alias="appId")
field_ids: list[int] = Field(default_factory=list, alias="fieldIds")
data_format: str = Field(default=DataFormat.Raw.name, alias="dataFormat")
page_number: int = Field(default=1, alias="pageNumber")
page_size: int = Field(default=50, alias="pageSize")
class QueryRecordsRequest(BaseModel):
"""Request parameters for querying records with a filter."""
model_config = ConfigDict(populate_by_name=True)
app_id: int = Field(alias="appId")
filter: str
field_ids: list[int] = Field(default_factory=list, alias="fieldIds")
data_format: str = Field(default=DataFormat.Raw.name, alias="dataFormat")
page_number: int = Field(default=1, alias="pageNumber")
page_size: int = Field(default=50, alias="pageSize")
class GetRecordByIdRequest(BaseModel):
"""Request parameters for fetching a record by ID."""
model_config = ConfigDict(populate_by_name=True)
app_id: int = Field(alias="appId")
record_id: int = Field(alias="recordId")
field_ids: list[int] = Field(default_factory=list, alias="fieldIds")
data_format: str = Field(default=DataFormat.Raw.name, alias="dataFormat")
class GetBatchRecordsRequest(BaseModel):
"""Request parameters for fetching multiple records by IDs."""
model_config = ConfigDict(populate_by_name=True)
app_id: int = Field(alias="appId")
record_ids: list[int] = Field(alias="recordIds")
field_ids: list[int] = Field(default_factory=list, alias="fieldIds")
data_format: str = Field(default=DataFormat.Raw.name, alias="dataFormat")
class DeleteBatchRecordsRequest(BaseModel):
"""Request payload for deleting multiple records."""
model_config = ConfigDict(populate_by_name=True)
app_id: int = Field(alias="appId")
record_ids: list[int] = Field(alias="recordIds")
class GetRecordsResponse(BaseModel):
"""Paginated response containing records."""
model_config = ConfigDict(populate_by_name=True)
page_number: int = Field(alias="pageNumber")
page_size: int = Field(alias="pageSize")
total_pages: int = Field(alias="totalPages")
total_records: int = Field(alias="totalRecords")
records: list[Record] = Field(alias="items")
class GetBatchRecordsResponse(BaseModel):
"""Response containing records for requested IDs."""
model_config = ConfigDict(populate_by_name=True)
count: int
records: list[Record] = Field(alias="items")
class AddOrUpdateRecordResponse(BaseModel):
"""Response containing the ID and warnings from an add/update operation."""
id: int
warnings: list[str] = Field(default_factory=list)
src/onspring_api_sdk/models/report.py
+56
View File
@@ -0,0 +1,56 @@
"""Pydantic models for Onspring report API requests and responses."""
from typing import Any, Optional
from pydantic import BaseModel, ConfigDict, Field
from onspring_api_sdk.enums import DataFormat, ReportDataType
class Row(BaseModel):
"""A single row of report data."""
model_config = ConfigDict(populate_by_name=True)
record_id: Optional[int] = Field(alias="recordId", default=None)
cells: list[Any]
class Report(BaseModel):
"""Represents an Onspring report."""
model_config = ConfigDict(populate_by_name=True)
app_id: int = Field(alias="appId")
id: int
name: str
description: Optional[str] = None
class GetReportByIdRequest(BaseModel):
"""Request parameters for fetching a report by ID."""
model_config = ConfigDict(populate_by_name=True)
report_id: int = Field(alias="reportId")
api_data_format: str = Field(default=DataFormat.Raw.name, alias="apiDataFormat")
data_type: str = Field(default=ReportDataType.ReportData.name, alias="dataType")
class GetReportByIdResponse(BaseModel):
"""Response containing report columns and rows."""
columns: list[str]
rows: list[Row]
class GetReportsByAppIdResponse(BaseModel):
"""Paginated response containing reports for an app."""
model_config = ConfigDict(populate_by_name=True)
page_number: int = Field(alias="pageNumber")
page_size: int = Field(alias="pageSize")
total_pages: int = Field(alias="totalPages")
total_records: int = Field(alias="totalRecords")
reports: list[Report] = Field(alias="items")