| Name | telepact JSON |
| Version |
1.0.0a164
JSON |
| download |
| home_page | None |
| Summary | The Python Telepact library |
| upload_time | 2025-10-28 13:39:54 |
| maintainer | None |
| docs_url | None |
| author | Telepact Authors |
| requires_python | >=3.11 |
| license | Apache-2.0 |
| keywords |
|
| VCS |
|
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
## Telepact Library for Python
### Installation
```
pip install telepact
```
### Usage
API:
```json
[
{
"fn.greet": {
"subject": "string"
},
"->": {
"Ok_": {
"message": "string"
}
}
}
]
```
Server:
```py
files = TelepactSchemaFiles('/directory/containing/api/files')
schema = TelepactSchema.from_file_json_map(files.filenames_to_json)
async def handler(request_message: 'Message') -> 'Message':
function_name = request_message.body.keys[0]
arguments = request_message.body[function_name]
try:
# Early in the handler, perform any pre-flight "middleware" operations, such as
# authentication, tracing, or logging.
log.info("Function started", {'function': function_name})
# Dispatch request to appropriate function handling code.
# (This example uses manual dispatching, but you can also use a more advanced pattern.)
if function_name == 'fn.greet':
subject = arguments['subject']
return Message({}, {'Ok_': {'message': f'Hello {subject}!'}})
raise Exception('Function not found')
finally:
# At the end the handler, perform any post-flight "middleware" operations
log.info("Function finished", {'function': function_name})
options = Server.Options()
server = Server(schema, handler, options)
# Wire up request/response bytes from your transport of choice
async def transport_handler(request_bytes: bytes) -> bytes:
response = await server.process(request_bytes)
return response.bytes
transport.receive(transport_handler)
```
Client:
```py
async def adapter(m: Message, s: Serializer) -> Message:
request_bytes = s.serialize(m)
# Wire up request/response bytes to your transport of choice
response_bytes = await transport.send(request_bytes)
return s.deserialize(response_bytes)
options = Client.Options()
client = Client(adapter, options)
# Inside your async application code:
request = Message({}, {'fn.greet': {'subject': 'World'}})
response = await client.request(request)
if response.get_body_target() == 'Ok_':
ok_payload = response.get_body_payload()
print(ok_payload['message'])
else:
raise RuntimeError(f"Unexpected response: {response.body}")
```
For more concrete usage examples,
[see the tests](https://github.com/Telepact/telepact/blob/main/test/lib/py/telepact_test/test_server.py).
Raw data
{
"_id": null,
"home_page": null,
"name": "telepact",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.11",
"maintainer_email": null,
"keywords": null,
"author": "Telepact Authors",
"author_email": "telepact@googlegroups.com",
"download_url": "https://files.pythonhosted.org/packages/b2/2c/cdfa24b3736d2bc41d0290f8f798917ed39c69975004d12fd9f892f90790/telepact-1.0.0a164.tar.gz",
"platform": null,
"description": "## Telepact Library for Python\n\n### Installation\n\n```\npip install telepact\n```\n\n### Usage\n\nAPI:\n\n```json\n[\n {\n \"fn.greet\": {\n \"subject\": \"string\"\n },\n \"->\": {\n \"Ok_\": {\n \"message\": \"string\"\n }\n }\n }\n]\n```\n\nServer:\n\n```py\n\nfiles = TelepactSchemaFiles('/directory/containing/api/files')\nschema = TelepactSchema.from_file_json_map(files.filenames_to_json)\n\nasync def handler(request_message: 'Message') -> 'Message':\n function_name = request_message.body.keys[0]\n arguments = request_message.body[function_name]\n\n try:\n # Early in the handler, perform any pre-flight \"middleware\" operations, such as\n # authentication, tracing, or logging.\n log.info(\"Function started\", {'function': function_name})\n\n # Dispatch request to appropriate function handling code.\n # (This example uses manual dispatching, but you can also use a more advanced pattern.)\n if function_name == 'fn.greet':\n subject = arguments['subject']\n return Message({}, {'Ok_': {'message': f'Hello {subject}!'}})\n\n raise Exception('Function not found')\n finally:\n # At the end the handler, perform any post-flight \"middleware\" operations\n log.info(\"Function finished\", {'function': function_name})\n\n\noptions = Server.Options()\nserver = Server(schema, handler, options)\n\n\n# Wire up request/response bytes from your transport of choice\nasync def transport_handler(request_bytes: bytes) -> bytes:\n response = await server.process(request_bytes)\n return response.bytes\n\ntransport.receive(transport_handler)\n```\n\nClient:\n\n```py\nasync def adapter(m: Message, s: Serializer) -> Message:\n request_bytes = s.serialize(m)\n\n # Wire up request/response bytes to your transport of choice\n response_bytes = await transport.send(request_bytes)\n\n return s.deserialize(response_bytes)\n\noptions = Client.Options()\nclient = Client(adapter, options)\n\n# Inside your async application code:\nrequest = Message({}, {'fn.greet': {'subject': 'World'}})\nresponse = await client.request(request)\nif response.get_body_target() == 'Ok_':\n ok_payload = response.get_body_payload()\n print(ok_payload['message'])\nelse:\n raise RuntimeError(f\"Unexpected response: {response.body}\")\n```\n\nFor more concrete usage examples,\n[see the tests](https://github.com/Telepact/telepact/blob/main/test/lib/py/telepact_test/test_server.py).\n",
"bugtrack_url": null,
"license": "Apache-2.0",
"summary": "The Python Telepact library",
"version": "1.0.0a164",
"project_urls": null,
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "1d1ca48227bd7f54d3132f01365d8de27d7ca1bd7d3bd308a83e3e11e478cb49",
"md5": "f0873633cecc85b4410152999a5d8e62",
"sha256": "5e7b28328d7cae674da00cfe17034e71535874509631e6a54735b4a0b0806a70"
},
"downloads": -1,
"filename": "telepact-1.0.0a164-py3-none-any.whl",
"has_sig": false,
"md5_digest": "f0873633cecc85b4410152999a5d8e62",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.11",
"size": 159466,
"upload_time": "2025-10-28T13:39:52",
"upload_time_iso_8601": "2025-10-28T13:39:52.715025Z",
"url": "https://files.pythonhosted.org/packages/1d/1c/a48227bd7f54d3132f01365d8de27d7ca1bd7d3bd308a83e3e11e478cb49/telepact-1.0.0a164-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "b22ccdfa24b3736d2bc41d0290f8f798917ed39c69975004d12fd9f892f90790",
"md5": "ee1456f5cde4b3be9f12ab60d6e99347",
"sha256": "73aa8bc3d82b6b9c295d2d3f87f2e61781a2987e89685ea6cf1cdfc7647fcdb8"
},
"downloads": -1,
"filename": "telepact-1.0.0a164.tar.gz",
"has_sig": false,
"md5_digest": "ee1456f5cde4b3be9f12ab60d6e99347",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.11",
"size": 46111,
"upload_time": "2025-10-28T13:39:54",
"upload_time_iso_8601": "2025-10-28T13:39:54.236448Z",
"url": "https://files.pythonhosted.org/packages/b2/2c/cdfa24b3736d2bc41d0290f8f798917ed39c69975004d12fd9f892f90790/telepact-1.0.0a164.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-10-28 13:39:54",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "telepact"
}
