# 7TV EventAPI Wrapper
[](https://dc.yoggies.dev/)
[](https://www.python.org/downloads/release/python-380/)
[](https://pypi.org/project/eventapi/)
[](https://pypi.org/project/eventapi/)
This is a Python wrapper for the 7TV EventAPI, which provides async access to the websocket events on
their [Emote Platform](https://7tv.app).
## Requirements
- Python 3.8 or higher
## Installation
You can install the 7TV EventAPI Wrapper using pip. Open your terminal and run the following command:
```shell
pip install eventapi
```
## Usage
> Here's a simple example of how to use wrapper:
```python
import asyncio
from eventapi.EventApi import EventApi
from eventapi.WebSocket import (
EventType,
ResponseTypes,
SubscriptionCondition,
SubscriptionData,
)
async def callback(data: ResponseTypes):
print("Message from callback: ", data)
async def main():
# create instance of EventApi with message callback
app = EventApi(callback=callback)
# connect to websocket
await app.connect()
print("Connected")
# create subscription with specified condition
condition = SubscriptionCondition(object_id="6433b7cec07d26f890dd2d01")
subscription = SubscriptionData(
subscription_type=EventType.EMOTE_SET_ALL, condition=condition
)
await app.subscribe(subscription_data=subscription)
# you can also use async iterator without specifying callback
async for message in app:
print("Message from async iterator: ", message)
# run forever if we are not using async for
await asyncio.Future()
asyncio.run(main())
```
<hr>
> Discord.py / Pycord example:
<img src="https://github.com/yoggys/eventapi/blob/master/assets/example_dc.png" alt="Discord Bot example" height="450px">
```python
from typing import Any, Dict, List
import aiohttp
import discord
from eventapi.EventApi import EventApi
from eventapi.WebSocket import (
Dispatch,
EventType,
ResponseTypes,
SubscriptionCondition,
SubscriptionData,
)
async def format_url(data: Dict[str, Any]) -> str:
base_url = "https://cdn.7tv.app/emote/{}/4x".format(data.get("id"))
if "animated" not in data:
async with aiohttp.ClientSession() as cs:
async with cs.get(base_url + ".gif") as r:
if r.status == 200:
base_url += ".gif"
else:
if data.get("animated"):
base_url += ".gif"
return base_url
async def callback(data: ResponseTypes) -> None:
if not isinstance(data, Dispatch):
return
if data.type != EventType.EMOTE_SET_UPDATE:
return
channel = client.get_channel(927288026000945162) # your channel id
changes: List[discord.Embed] = []
def add_change(description: str, color: discord.Color, image_url: str):
changes.append(
discord.Embed(description=description, color=color).set_image(url=image_url)
)
if data.body.pulled:
for pulled in data.body.pulled:
add_change(
"**Deleted emote:** `{}`".format(pulled.old_value.get("name")),
discord.Color.brand_red(),
await format_url(pulled.old_value),
)
if data.body.pushed:
for pushed in data.body.pushed:
add_change(
"**Added emote:** `{}`".format(pushed.value.get("name")),
discord.Color.brand_green(),
await format_url(pushed.value.get("data")),
)
if data.body.updated:
for updated in data.body.updated:
add_change(
"**Edited emote:** `{}` » `{}`".format(
updated.old_value.get("name"), updated.value.get("name")
),
discord.Color.yellow(),
await format_url(updated.value),
)
if len(changes) > 0:
await channel.send(embeds=changes)
intents = discord.Intents.default()
client = discord.Client(intents=intents)
client.eventapi = EventApi(callback=callback)
@client.listen("on_ready", once=True)
async def on_ready():
await client.eventapi.connect()
condition = SubscriptionCondition(
object_id="6433b7cec07d26f890dd2d01"
) # your emote-set id
subscription = SubscriptionData(
subscription_type=EventType.EMOTE_SET_ALL, condition=condition
)
await client.eventapi.subscribe(subscription_data=subscription)
client.run("your token here") # your token
```
## Contributing
Contributions are welcome! If you encounter any issues or have suggestions for improvements, please open an issue or
submit a pull request on the [GitHub repository](https://github.com/yoggys/eventapi).
## License
This project is licensed under the [MIT License](https://opensource.org/licenses/MIT).
Raw data
{
"_id": null,
"home_page": "https://github.com/yoggys/eventapi",
"name": "eventapi",
"maintainer": null,
"docs_url": null,
"requires_python": null,
"maintainer_email": null,
"keywords": "python, 7tv, eventapi, 7tvapi, 7tveventapi, async, asyncio, wrapper",
"author": "yoggys",
"author_email": "yoggies@yoggies.dev",
"download_url": "https://files.pythonhosted.org/packages/49/e4/853f0babd557085d77e584db560362ff5ebfba65cdc1070a63bcb2ec7b59/eventapi-1.1.2.tar.gz",
"platform": null,
"description": "# 7TV EventAPI Wrapper\r\n\r\n[](https://dc.yoggies.dev/)\r\n[](https://www.python.org/downloads/release/python-380/)\r\n[](https://pypi.org/project/eventapi/)\r\n[](https://pypi.org/project/eventapi/)\r\n\r\nThis is a Python wrapper for the 7TV EventAPI, which provides async access to the websocket events on\r\ntheir [Emote Platform](https://7tv.app).\r\n\r\n## Requirements\r\n\r\n- Python 3.8 or higher\r\n\r\n## Installation\r\n\r\nYou can install the 7TV EventAPI Wrapper using pip. Open your terminal and run the following command:\r\n\r\n```shell\r\npip install eventapi\r\n```\r\n\r\n## Usage\r\n\r\n> Here's a simple example of how to use wrapper:\r\n\r\n```python\r\nimport asyncio\r\n\r\nfrom eventapi.EventApi import EventApi\r\nfrom eventapi.WebSocket import (\r\n EventType,\r\n ResponseTypes,\r\n SubscriptionCondition,\r\n SubscriptionData,\r\n)\r\n\r\n\r\nasync def callback(data: ResponseTypes):\r\n print(\"Message from callback: \", data)\r\n\r\n\r\nasync def main():\r\n # create instance of EventApi with message callback\r\n app = EventApi(callback=callback)\r\n\r\n # connect to websocket\r\n await app.connect()\r\n print(\"Connected\")\r\n\r\n # create subscription with specified condition\r\n condition = SubscriptionCondition(object_id=\"6433b7cec07d26f890dd2d01\")\r\n subscription = SubscriptionData(\r\n subscription_type=EventType.EMOTE_SET_ALL, condition=condition\r\n )\r\n await app.subscribe(subscription_data=subscription)\r\n\r\n # you can also use async iterator without specifying callback\r\n async for message in app:\r\n print(\"Message from async iterator: \", message)\r\n\r\n # run forever if we are not using async for\r\n await asyncio.Future()\r\n\r\n\r\nasyncio.run(main())\r\n```\r\n\r\n<hr>\r\n\r\n> Discord.py / Pycord example:\r\n\r\n<img src=\"https://github.com/yoggys/eventapi/blob/master/assets/example_dc.png\" alt=\"Discord Bot example\" height=\"450px\">\r\n\r\n```python\r\nfrom typing import Any, Dict, List\r\n\r\nimport aiohttp\r\nimport discord\r\n\r\nfrom eventapi.EventApi import EventApi\r\nfrom eventapi.WebSocket import (\r\n Dispatch,\r\n EventType,\r\n ResponseTypes,\r\n SubscriptionCondition,\r\n SubscriptionData,\r\n)\r\n\r\n\r\nasync def format_url(data: Dict[str, Any]) -> str:\r\n base_url = \"https://cdn.7tv.app/emote/{}/4x\".format(data.get(\"id\"))\r\n if \"animated\" not in data:\r\n async with aiohttp.ClientSession() as cs:\r\n async with cs.get(base_url + \".gif\") as r:\r\n if r.status == 200:\r\n base_url += \".gif\"\r\n else:\r\n if data.get(\"animated\"):\r\n base_url += \".gif\"\r\n return base_url\r\n\r\n\r\nasync def callback(data: ResponseTypes) -> None:\r\n if not isinstance(data, Dispatch):\r\n return\r\n if data.type != EventType.EMOTE_SET_UPDATE:\r\n return\r\n\r\n channel = client.get_channel(927288026000945162) # your channel id\r\n changes: List[discord.Embed] = []\r\n\r\n def add_change(description: str, color: discord.Color, image_url: str):\r\n changes.append(\r\n discord.Embed(description=description, color=color).set_image(url=image_url)\r\n )\r\n\r\n if data.body.pulled:\r\n for pulled in data.body.pulled:\r\n add_change(\r\n \"**Deleted emote:** `{}`\".format(pulled.old_value.get(\"name\")),\r\n discord.Color.brand_red(),\r\n await format_url(pulled.old_value),\r\n )\r\n if data.body.pushed:\r\n for pushed in data.body.pushed:\r\n add_change(\r\n \"**Added emote:** `{}`\".format(pushed.value.get(\"name\")),\r\n discord.Color.brand_green(),\r\n await format_url(pushed.value.get(\"data\")),\r\n )\r\n if data.body.updated:\r\n for updated in data.body.updated:\r\n add_change(\r\n \"**Edited emote:** `{}` \u00c2\u00bb `{}`\".format(\r\n updated.old_value.get(\"name\"), updated.value.get(\"name\")\r\n ),\r\n discord.Color.yellow(),\r\n await format_url(updated.value),\r\n )\r\n\r\n if len(changes) > 0:\r\n await channel.send(embeds=changes)\r\n\r\n\r\nintents = discord.Intents.default()\r\nclient = discord.Client(intents=intents)\r\nclient.eventapi = EventApi(callback=callback)\r\n\r\n\r\n@client.listen(\"on_ready\", once=True)\r\nasync def on_ready():\r\n await client.eventapi.connect()\r\n condition = SubscriptionCondition(\r\n object_id=\"6433b7cec07d26f890dd2d01\"\r\n ) # your emote-set id\r\n subscription = SubscriptionData(\r\n subscription_type=EventType.EMOTE_SET_ALL, condition=condition\r\n )\r\n await client.eventapi.subscribe(subscription_data=subscription)\r\n\r\n\r\nclient.run(\"your token here\") # your token\r\n```\r\n\r\n\r\n\r\n## Contributing\r\n\r\nContributions are welcome! If you encounter any issues or have suggestions for improvements, please open an issue or\r\nsubmit a pull request on the [GitHub repository](https://github.com/yoggys/eventapi).\r\n\r\n## License\r\n\r\nThis project is licensed under the [MIT License](https://opensource.org/licenses/MIT).\r\n",
"bugtrack_url": null,
"license": null,
"summary": "Wrapper for 7TV EventAPI.",
"version": "1.1.2",
"project_urls": {
"Homepage": "https://github.com/yoggys/eventapi"
},
"split_keywords": [
"python",
" 7tv",
" eventapi",
" 7tvapi",
" 7tveventapi",
" async",
" asyncio",
" wrapper"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "0c8618fead10aaadb3d42d72792c97dbabe3f08afa8d2c880f5e5ed4ced7d9fb",
"md5": "ba59c70b5d4194350082718c3638a776",
"sha256": "7c6441743ebad38bb43d2a9a8a6e87ab382e3adb89cef6b36f083c10c6aedd6f"
},
"downloads": -1,
"filename": "eventapi-1.1.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "ba59c70b5d4194350082718c3638a776",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": null,
"size": 9044,
"upload_time": "2024-09-16T10:11:43",
"upload_time_iso_8601": "2024-09-16T10:11:43.656081Z",
"url": "https://files.pythonhosted.org/packages/0c/86/18fead10aaadb3d42d72792c97dbabe3f08afa8d2c880f5e5ed4ced7d9fb/eventapi-1.1.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "49e4853f0babd557085d77e584db560362ff5ebfba65cdc1070a63bcb2ec7b59",
"md5": "ebfea8347c8442f2d066c095c326e851",
"sha256": "a43411b7b6c85e04524d642f4745ba6c54097d80772072ccc43a1875a2185810"
},
"downloads": -1,
"filename": "eventapi-1.1.2.tar.gz",
"has_sig": false,
"md5_digest": "ebfea8347c8442f2d066c095c326e851",
"packagetype": "sdist",
"python_version": "source",
"requires_python": null,
"size": 10069,
"upload_time": "2024-09-16T10:11:46",
"upload_time_iso_8601": "2024-09-16T10:11:46.467864Z",
"url": "https://files.pythonhosted.org/packages/49/e4/853f0babd557085d77e584db560362ff5ebfba65cdc1070a63bcb2ec7b59/eventapi-1.1.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-09-16 10:11:46",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "yoggys",
"github_project": "eventapi",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "eventapi"
}
JSON
