Skip to content

Add fetching additional DataSource fields #122

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
wscourge merged 5 commits into from
Dec 17, 2025
Merged

Add fetching additional DataSource fields #122

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
49afff3
Add fetching additional DataSource fields
wscourge Dec 1, 2025
5d67fed
Update auto_churn_subscription_setting response format
wscourge Dec 9, 2025
c2bb8e0
Update python bool Trues to be converted to strings
wscourge Dec 11, 2025
dfd09e5
Fix flake8 violation
wscourge Dec 11, 2025
b7f94c9
Update invoice_handling_setting format to fully backwards-compatible
wscourge Dec 15, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 13 additions & 2 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,8 +110,19 @@ Available methods in Import API:

```python
chartmogul.DataSource.create(config, data={'name': 'In-house billing'})
chartmogul.DataSource.retrieve(config, uuid='ds_5915ee5a-babd-406b-b8ce-d207133fb4cb')
chartmogul.DataSource.all(config)
chartmogul.DataSource.retrieve(
config,
uuid='ds_5915ee5a-babd-406b-b8ce-d207133fb4cb',
with_processing_status=True,
with_auto_churn_subscription_setting=True,
with_invoice_handling_setting=True
)
chartmogul.DataSource.all(
config,
with_processing_status=True,
with_auto_churn_subscription_setting=True,
with_invoice_handling_setting=True
)
chartmogul.DataSource.destroy(config, uuid='ds_5915ee5a-babd-406b-b8ce-d207133fb4cb')
```

Expand Down
50 changes: 48 additions & 2 deletions chartmogul/api/data_source.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,23 +1,69 @@
from marshmallow import Schema, fields, post_load, EXCLUDE
from ..resource import Resource
from ..resource import Resource, DataObject
from collections import namedtuple


class ProcessingStatus(DataObject):
class _Schema(Schema):
processed = fields.Integer(allow_none=True)
pending = fields.Integer(allow_none=True)
failed = fields.Integer(allow_none=True)

@post_load
def make(self, data, **kwargs):
return ProcessingStatus(**data)


class AutoChurnSubscriptionSetting(DataObject):
class _Schema(Schema):
enabled = fields.Boolean()
interval = fields.Integer(allow_none=True)

@post_load
def make(self, data, **kwargs):
return AutoChurnSubscriptionSetting(**data)


class DataSource(Resource):
"""
https://dev.chartmogul.com/v1.0/reference#data-sources
"""

_path = "/data_sources{/uuid}"
_root_key = "data_sources"
_many = namedtuple("DataSources", [_root_key])
_bool_query_params = [
'with_processing_status',
'with_auto_churn_subscription_setting',
'with_invoice_handling_setting'
]
_many = namedtuple(
"DataSources",
[_root_key] + _bool_query_params,
defaults=[None, None, None]
)

@classmethod
def _preProcessParams(cls, params):
params = super()._preProcessParams(params)

for query_param in cls._bool_query_params:
if query_param in params and isinstance(params[query_param], bool):
if params[query_param] is True:
params[query_param] = 'true'
else:
del params[query_param]

return params
Copy link

@loomchild loomchild Dec 11, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[for later] This code is pretty generic, so we could move it to Resource.

Maybe even there's no need for bool_query_params variable and we can rely only on type, but this would be a bit riskier, since you'd have to assure that other resources and params like has_more still work.


class _Schema(Schema):
uuid = fields.String()
name = fields.String()
created_at = fields.DateTime()
status = fields.Str()
system = fields.Str()
processing_status = fields.Nested(ProcessingStatus._Schema, many=False, allow_none=True)
auto_churn_subscription_setting = fields.Nested(AutoChurnSubscriptionSetting._Schema, many=False, allow_none=True)
invoice_handling_setting = fields.Raw(allow_none=True)

@post_load
def make(self, data, **kwargs):
Expand Down
145 changes: 136 additions & 9 deletions test/api/test_data_source.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ def test_create(self, mock_requests):
"name": "test",
"uuid": "my_uuid",
"created_at": "2016-01-10T15:34:05.144Z",
"status": "never_imported",
"status": "idle",
},
)

Expand All @@ -47,27 +47,90 @@ def test_retrieve(self, mock_requests):
"name": "test",
"uuid": "my_uuid",
"created_at": "2016-01-10T15:34:05Z",
"status": "never_imported",
"status": "idle",
"processing_status": {
"processed": 61,
"failed": 3,
"pending": 8
},
"auto_churn_subscription_setting": {
"enabled": True,
"interval": 30
},
"invoice_handling_setting": {
"manual": {
"create_subscription_when_invoice_is": "open",
"update_subscription_when_invoice_is": "open",
"prevent_subscription_for_invoice_voided": True,
"prevent_subscription_for_invoice_refunded": False,
"prevent_subscription_for_invoice_written_off": True
},
"automatic": {
"create_subscription_when_invoice_is": "open",
"update_subscription_when_invoice_is": "open",
"prevent_subscription_for_invoice_voided": True,
"prevent_subscription_for_invoice_refunded": False,
"prevent_subscription_for_invoice_written_off": True
}
}
},
)

config = Config("token")
ds = DataSource.retrieve(config, uuid="my_uuid").get()
ds = DataSource.retrieve(
config,
uuid="my_uuid",
with_processing_status=True,
with_auto_churn_subscription_setting=True,
with_invoice_handling_setting=True
).get()
expected = DataSource(
**{
"name": "test",
"uuid": "my_uuid",
"created_at": datetime(2016, 1, 10, 15, 34, 5),
"status": "never_imported",
"status": "idle",
"processing_status": {
"processed": 61,
"failed": 3,
"pending": 8
},
"auto_churn_subscription_setting": {
"enabled": True,
"interval": 30
},
"invoice_handling_setting": {
"manual": {
"create_subscription_when_invoice_is": "open",
"update_subscription_when_invoice_is": "open",
"prevent_subscription_for_invoice_voided": True,
"prevent_subscription_for_invoice_refunded": False,
"prevent_subscription_for_invoice_written_off": True
},
"automatic": {
"create_subscription_when_invoice_is": "open",
"update_subscription_when_invoice_is": "open",
"prevent_subscription_for_invoice_voided": True,
"prevent_subscription_for_invoice_refunded": False,
"prevent_subscription_for_invoice_written_off": True
}
}
}
)

self.assertEqual(mock_requests.call_count, 1, "expected call")
self.assertEqual(mock_requests.last_request.qs, {})
self.assertEqual(mock_requests.last_request.qs, {
"with_processing_status": ["true"],
"with_auto_churn_subscription_setting": ["true"],
"with_invoice_handling_setting": ["true"]
})
self.assertEqual(mock_requests.last_request.text, None)
self.assertTrue(isinstance(ds, DataSource))
self.assertTrue(isinstance(ds.created_at, datetime))
self.assertEqual(ds.name, "test")
self.assertEqual(ds.processing_status.processed, 61)
self.assertEqual(ds.auto_churn_subscription_setting.interval, 30)
self.assertEqual(ds.invoice_handling_setting['manual']['create_subscription_when_invoice_is'], "open")

@requests_mock.mock()
def test_all(self, mock_requests):
Expand All @@ -81,31 +144,95 @@ def test_all(self, mock_requests):
"name": "test",
"uuid": "my_uuid",
"created_at": "2016-01-10T15:34:05Z",
"status": "never_imported",
"status": "idle",
"processing_status": {
"processed": 61,
"failed": 3,
"pending": 8
},
"auto_churn_subscription_setting": {
"enabled": True,
"interval": 30
},
"invoice_handling_setting": {
"manual": {
"create_subscription_when_invoice_is": "open",
"update_subscription_when_invoice_is": "open",
"prevent_subscription_for_invoice_voided": True,
"prevent_subscription_for_invoice_refunded": False,
"prevent_subscription_for_invoice_written_off": True
},
"automatic": {
"create_subscription_when_invoice_is": "open",
"update_subscription_when_invoice_is": "open",
"prevent_subscription_for_invoice_voided": True,
"prevent_subscription_for_invoice_refunded": False,
"prevent_subscription_for_invoice_written_off": True
}
}
}
]
},
)

config = Config("token")
ds = DataSource.all(config).get()
ds = DataSource.all(
config,
with_processing_status=True,
with_auto_churn_subscription_setting=True,
with_invoice_handling_setting=True
).get()
expected = DataSource._many(
data_sources=[
DataSource(
**{
"name": "test",
"uuid": "my_uuid",
"created_at": datetime(2016, 1, 10, 15, 34, 5),
"status": "never_imported",
"status": "idle",
"processing_status": {
"processed": 61,
"failed": 3,
"pending": 8
},
"auto_churn_subscription_setting": {
"enabled": True,
"interval": 30
},
"invoice_handling_setting": {
"manual": {
"create_subscription_when_invoice_is": "open",
"update_subscription_when_invoice_is": "open",
"prevent_subscription_for_invoice_voided": True,
"prevent_subscription_for_invoice_refunded": False,
"prevent_subscription_for_invoice_written_off": True
},
"automatic": {
"create_subscription_when_invoice_is": "open",
"update_subscription_when_invoice_is": "open",
"prevent_subscription_for_invoice_voided": True,
"prevent_subscription_for_invoice_refunded": False,
"prevent_subscription_for_invoice_written_off": True
}
}
}
)
]
)

self.assertEqual(mock_requests.call_count, 1, "expected call")
self.assertEqual(mock_requests.last_request.qs, {})
self.assertEqual(mock_requests.last_request.qs, {
"with_processing_status": ["true"],
"with_auto_churn_subscription_setting": ["true"],
"with_invoice_handling_setting": ["true"]
})
self.assertEqual(mock_requests.last_request.text, None)
self.assertTrue(isinstance(ds.data_sources[0], DataSource))
self.assertTrue(isinstance(ds.data_sources[0].created_at, datetime))
self.assertEqual(ds.data_sources[0].name, "test")
self.assertEqual(ds.data_sources[0].processing_status.processed, 61)
self.assertEqual(ds.data_sources[0].auto_churn_subscription_setting.interval, 30)
self.assertEqual(ds.data_sources[0].invoice_handling_setting['manual']['create_subscription_when_invoice_is'], "open")

@requests_mock.mock()
def test_destroy(self, mock_requests):
Expand Down