Для начала: кто я? Меня зовут Костя, являюсь бекенд разработчиком, питонистом. 20 годиков, в общем еще есть время чтобы поработать над open source

Сначала небольшое введение, почему я решился(а точнее почему я сделал) библиотеку Fastsio.

Большинство из нас пишут обычное REST API, иногда используя SSE или websockets(да еще есть graphQL и grpc из популярного). Но, есть еще socket.io, это условные аналоги вебсокетов с блек джеком и ш.. удобным сахарком(тут я имею ввиду, что каждое сообщение имеет event, и так же клиенты имеют неймспейсы). Почему это условные аналоги? Да потому что по сути socket.io это протокол поверх websockets(и не только websockets, там и webTransport, и HTTP Pooling есть), но они не взаимозаменяемы. Мы сейчас не будет углубляться как работает эта технология, мы тут занимаемся чуть-чуть другим, а именно удобной библиотекой для работы с socket.io.

Кто знаком с этим творением, тот думаю и знаком с python-socketio библиотекой. Кто не знаком, сейчас обьясню что мне в ней НЕ нравится

1. Нет удобных аннотаций.
   Кто пишет на FastAPI, Aiogram, FastStream, и других библиотеках/фреймворках где просто нельзя писать без аннотаций, поймет эту боль. Меня очень сильно это бесило

2. Нет роутов
   Вот просто не можете без костылей это сделать, ну нельзя. Внутри обработчиков вам нужен объект сервера. А обработчики нужно привязать к серверу, а сервер к обработчику. В общем возникает циклическая зависимость. Грустно.

   

3. Нет middlewares
   Отдельная боль, хоть пока мне и не нужна была. Но решил что все таки в такой библиотеке она должна быть

4. Нет автодокументации(asyncapi)
   Справедливости ради это есть в сторонних либах, но в них нет всего остального. В общем шило на мыло

5. Нет интеграции с pydantiс
   Скажите удобно когда автоматически происходит десериализация в модельку? Вот и мне так кажется. И даже не обязательно Pydantic, есть Packet, MsgPack с ними тоже надо бы интегрироваться.

Вот после этих недостающих я решил, что пора сжать волю в кулак и войти в этот дивный мир open source. Название я выбирал случайно, во время мозгового штурма попалось интервью про FastStream от Никиты Соболева и Никиты Пастухова. И так как я брал многое и от FastAPI, то решил что будет FastSIO.

И теперь опишу основные фичи и попробую продать вам мысль: ЗАЧЕМ использовать FastSIO заместо python-socketio.

1. Добавили аннотации для всех стандартных событий. Подключение, отключение, получение ивента. Все можно получить по необходимости

   import datetime
   
   from pydantic import BaseModel
   
   import fastsio
   from fastsio import SocketID, Environ, Auth, AsyncServer, Data, Reason
   
   sio = fastsio.AsyncServer(
       async_mode="asgi",
       cors_allowed_origins="*",
   )
   
   
   @sio.on("message.send")
   async def message__send(sid: SocketID, sio: AsyncServer, data: Data):
       await sio.emit("message.new", data=data, to=sid)
   
   @sio.on("message.edit")
   async def message__edit(sid: SocketID, sio: AsyncServer, data: Data):
       pass
   
   @sio.on("disconnect")
   async def disconnect(sid: SocketID, sio: AsyncServer, reason: Reason):
       print(f"client {sid} disconnect. Reason: {reason}")
   
   
   app = fastsio.ASGIApp(sio)

2. Так же добавил роутеры. Так как очень нравится такой подход(что в fastapi, что в aiogram)

   # app.py
   import fastsio
   from .routers import router
   
   sio = fastsio.AsyncServer(
       async_mode="asgi",
       cors_allowed_origins=None,
   )
   
   # added all routers
   sio.add_router(router=router)
   
   
   
   # router.py
   from fastsio import RouterSIO
   
   router = RouterSIO(namespace="/app")
   
   
   @router.on("connect", namespace="/room")  # override router namespace
   async def on_connect(sid: SocketID, environ: Environ):
       print("connect ", sid)
   
   
   @router.on("disconnect")
   async def on_disconnect(sid: SocketID):
       print("disconnect ", sid)
   
   
   @router.on("message")
   async def on_message(sid: SocketID, data: Data):
       print("message ", data)
   

   Мы можем собирать удобно приложения, и группировать их через роутеры :) В самих RouterSIO мы можем определить еще namespace, так что не надо по 20 раз писать одно и тоже для каждого обработчика

3. Дальше добавлял middlewares. Работают они как и везде(первый зашел, последний вышел), примеры создания можно почитать в документации

   from fastsio import AsyncServer
   
   sio = AsyncServer()
   
   # Add middleware with default settings
   sio.add_middleware(LoggingMiddleware())
   
   # Override middleware settings
   sio.add_middleware(
       AuthMiddleware(),
       events=["join_room", "send_message"],
       namespace="/chat"
   )
   
   # Global middleware
   sio.add_middleware(LoggingMiddleware(), global_middleware=True)

4. Ооо, любимое, автодокументация. Это и 5 пункт одновременно, так как автодокументация завязана на pydantic. С ней сейчас есть проблемы. Так как это асинхронное общение, то и собирать ивенты исходящие, это одно удовольствие(sarcasm). Так что пришлось костыль вводить в виде response_model

   import datetime
   
   from pydantic import BaseModel
   
   import fastsio
   from fastsio import SocketID, Environ, Auth, AsyncServer, Data, AsyncAPIConfig
   
   router = fastsio.RouterSIO()
   
   
   class DataMessage(BaseModel):
       session_id: str
       text: str
   
   
   class EditMessage(BaseModel):
       session_id: str
       text: str
       time_updated: datetime.datetime
   
   
   @router.on("message.send", response_model={"message.new": DataMessage})
   async def message__send(sid: SocketID, sio: AsyncServer, data: DataMessage):
       await sio.emit("message.new", data=data.model_dump(), to=sid)
   
   
   @router.on("message.edit")
   async def message__edit(sid: SocketID, sio: AsyncServer, data: DataMessage):
       pass
   
   
   sio = fastsio.AsyncServer(
       async_mode="asgi",
       cors_allowed_origins="*",
       # its default settings for config
       # asyncapi=AsyncAPIConfig(
       #     enabled=True,
       #     url="/asyncapi.json",
       #     expose_yaml=True,
       #     title="Socket.IO API",
       #     version="1.0.0",
       #     description=None,
       #     servers={},
       #     channel_prefix="",
       #     ui_url="/asyncapi",
       # ),
   )
   
   sio.add_router(router=router)
   
   app = fastsio.ASGIApp(sio)

   Ну в общем ничего нового для вас, стандартная pydantic модель, и стандартный подход для FastAPI. Прошу заметить, что доступная ui панель(по умолчанию /asyncapi), и пока что недоступны другие библиотеки кроме pydantic. Сейчас работаю над этим, чтобы бинарные данные можно было гонять.
   Так же, есть автовалидация моделей, удобно!

Это вроде бы основной дополнительный функционал для socket.io на питоне. Есть еще dependency injection в FastSIO, почитать можно тут. Но он мне не нравится как реализован, буду улучшать его.

Вот и весь обзор. Надеюсь, кому-нибудь облегчит жизнь библиотека.
Если есть вопросы/предложения/критика, то пишите в комментарии или в ТГ - cicwak5560