I'm working with an API, where the url can be constructed with setting_ids. To more clearly indicate the functionality of the setting ids I am mapping them to a WriterClass' methods, with relevant names.
In the code below these are method_a
to method_c
, in the true application this is a list of ~250 settings.
The functionality I desire would access the API for a desired setting through the structure writer_instance.setting_name(value)
- returning a to be awaited coroutine.
For this I'm using descriptors, with which I'm still familiarising myself. This is my main motivation for asking this question, to ensure I'm utilising them correctly and as efficiently as possible - and if not, to learn how to do so now and in the future.
Below is code that functionally does what I desire, with placeholder prints instead of code accessing the (private) API.
The Credentials class is a separate class as it's utilised by other Classes which also access the same API.
import asyncio
import httpx
DOMAIN = 'api.mysite.com'
class Credentials:
# In true context this fetches an api_token from database using the serial.
def __init__(self, device_serial: str, api_token):
self.device_serial = device_serial
self._api_token = api_token
self.headers = self.__build_headers()
def __build_headers(self):
headers = {
'Authorization': ('Bearer ' + self._api_token),
'Content-Type': 'application/json',
'Accept': 'application/json',
}
return headers
class MethodDescriptor:
def __init__(self, id):
self._id = id
def __get__(self, instance, owner):
# Using this so that the final method has access to instance variables.
# Not happy with this approach necessarily, but it gets the results I desire.
return BoundMethod(self._id, instance)
class BoundMethod:
def __init__(self, id, instance):
self._id = id
self._instance = instance
async def __call__(self, value):
# placeholder logic for actual API call.
await asyncio.sleep(1) # Simulate work
print(
f"Method called with id={self._id}, {value=}, at url={self._instance.domain} for serial {self._instance.device_serial}")
return f'accessed setting number {self._id}'
class WriterClass:
method_a = MethodDescriptor(1)
method_b = MethodDescriptor(2)
method_c = MethodDescriptor(3)
def __init__(self, credentials: Credentials, client: httpx.AsyncClient):
self.device_serial = credentials.device_serial
self.headers = credentials.headers
self.domain = DOMAIN
self.client = client
# Example usage
async def main():
creds = Credentials("123AA", "API_TOKEN") # API token obtained elsewhere in real code.
async with httpx.AsyncClient() as client:
writer = WriterClass(creds, client)
task1 = asyncio.create_task(writer.method_c("some_str")) # should use __call__ with self._id = 3, value="some_str" and instance vars.
task2 = asyncio.create_task(writer.method_b("12:34"))
task3 = asyncio.create_task(writer.method_a("3700"))
results = await asyncio.gather(task1, task2, task3)
print(results)
if __name__ == '__main__':
asyncio.run(main())
Which returns:
Method called with id=3, value='some_str', at url=api.mysite.com for serial 123AA
Method called with id=2, value='12:34', at url=api.mysite.com for serial 123AA
Method called with id=1, value='3700', at url=api.mysite.com for serial 123AA
['accessed setting number 3', 'accessed setting number 2', 'accessed setting number 1']
As expected.