
Security News
Deno 2.2 Improves Dependency Management and Expands Node.js Compatibility
Deno 2.2 enhances Node.js compatibility, improves dependency management, adds OpenTelemetry support, and expands linting and task automation for developers.
SIO-AsyncAPI is a Python library built on the top of Flask-SocketIO and driven by AsyncAPI. It allows you to generate an AsyncAPI specification from your SocketIO server and validate messages against it.
Similar to FastAPI, SIO-AsyncAPI allows you to define your SocketIO server using Python type annotations and Pydantic models. It also provides a way to generate an AsyncAPI specification from your SocketIO server.
pip install sio_asyncapi
# examples/simple.py
from flask import Flask
from sio_asyncapi import AsyncAPISocketIO, ResponseValidationError, RequestValidationError
from pydantic import BaseModel, Field, EmailStr
from typing import Optional
import logging
logger = logging.getLogger(__name__)
app = Flask(__name__)
socketio = AsyncAPISocketIO(
app,
validate=True,
generate_docs=True,
version="1.0.0",
title="Demo",
description="Demo Server",
server_url="http://localhost:5000",
server_name="DEMO_SIO",
)
class UserSignUpRequest(BaseModel):
"""Request model for user sign up"""
email: EmailStr = Field(..., description="User email", example="bob@gmail.com")
password: str = Field(..., description="User password", example="123456")
class UserSignUpResponse(BaseModel):
"""Response model for user sign up"""
success: bool = Field(True, description="Success status")
error: Optional[str] = Field( None, description="Error message if any",
example="Invalid request")
@socketio.on("user_sign_up", get_from_typehint=True)
def user_sign_up(request: UserSignUpRequest) -> UserSignUpResponse:
"""User sign up"""
_ = request
return UserSignUpResponse(success=True, error=None)
@socketio.on_error_default
def default_error_handler(e: Exception):
"""
Default error handler. It called if no other error handler defined.
Handles RequestValidationError and ResponseValidationError errors.
"""
if isinstance(e, RequestValidationError):
logger.error(f"Request validation error: {e}")
return UserSignUpResponse(error=str(e), success=False).json()
elif isinstance(e, ResponseValidationError):
logger.critical(f"Response validation error: {e}")
raise e
else:
logger.critical(f"Unknown error: {e}")
raise e
if __name__ == '__main__':
socketio.run(app, debug=True)
# import pathlib
# if __name__ == "__main__":
# path = pathlib.Path(__file__).parent / "simple.yml"
# doc_str = socketio.asyncapi_doc.get_yaml()
# with open(path, "w") as f:
# f.write(doc_str)
# print(doc_str)
Here is how validation error looks like in FireCamp:
In order to get the AsyncAPI specification from your SocketIO server instead of running the server, you can do the following:
import pathlib
if __name__ == "__main__":
path = pathlib.Path(__file__).parent / "simple.yml"
doc_str = socketio.asyncapi_doc.get_yaml()
with open(path, "w") as f:
f.write(doc_str)
print(doc_str)
Example of the AsyncAPI specification generated from the above example:
# examples/simple.yml
asyncapi: 2.5.0
channels:
/:
publish:
message:
oneOf:
- $ref: '#/components/messages/User_Sign_Up'
subscribe:
message:
oneOf: []
x-handlers:
disconnect: disconnect
components:
messages:
User_Sign_Up:
description: User sign up
name: user_sign_up
payload:
$ref: '#/components/schemas/UserSignUpRequest'
deprecated: false
x-ack:
description: Response model for user sign up
properties:
error:
description: Error message if any
example: Invalid request
title: Error
type: string
success:
default: true
description: Success status
title: Success
type: boolean
title: UserSignUpResponse
type: object
schemas:
NoSpec:
deprecated: false
description: Specification is not provided
UserSignUpRequest:
description: Request model for user sign up
properties:
email:
description: User email
example: bob@gmail.com
format: email
title: Email
type: string
password:
description: User password
example: '123456'
title: Password
type: string
required:
- email
- password
title: UserSignUpRequest
type: object
UserSignUpResponse:
description: Response model for user sign up
properties:
error:
description: Error message if any
example: Invalid request
title: Error
type: string
success:
default: true
description: Success status
title: Success
type: boolean
title: UserSignUpResponse
type: object
info:
description: 'Demo Server
<br/> AsyncAPI currently does not support Socket.IO binding and Web Socket like
syntax used for now.
In order to add support for Socket.IO ACK value, AsyncAPI is extended with with
x-ack keyword.
This documentation should **NOT** be used for generating code due to these limitations.
'
title: Demo
version: 1.0.0
servers:
DEMO_SIO:
protocol: socketio
url: http://localhost:5000
Rendered version of the above AsyncAPI specification:
SIO-AsyncAPI is built on top of Flask-SocketIO and all unit tests of Flask-SocketIO are tested against SIO-AsyncAPI. If you converting your SocketIO server from Flask-SocketIO to SIO-AsyncAPI, you can be sure that your SocketIO server will work as expected. When converting your SocketIO server from Flask-SocketIO to SIO-AsyncAPI, it's as simple as changing the import statement:
# instead of `from flask_socketio import SocketIO`
from sio_asyncapi import AsyncAPISocketIO as SocketIO
...
# There are additional arguments that you can pass to the constructor of AsyncAPISocketIO
socketio = SocketIO(app)
...
Most of the implementation follows research done by Dimitrios Dedoussis (https://www.asyncapi.com/blog/socketio-part2) and uses some Pydantic models from here
SIO-AsyncAPI is still in its early stages and there are some features that are not yet implemented. If you are interested in contributing to SIO-AsyncAPI any contribution is welcome. Here is the list of missing features:
emit
messagesnamespaces
and rooms
connect
and disconnect
handlers auto documentationFAQs
Flask SocketIO with auto-generate Asyncapi documentation
We found that sio-asyncapi demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
Deno 2.2 enhances Node.js compatibility, improves dependency management, adds OpenTelemetry support, and expands linting and task automation for developers.
Security News
React's CRA deprecation announcement sparked community criticism over framework recommendations, leading to quick updates acknowledging build tools like Vite as valid alternatives.
Security News
Ransomware payment rates hit an all-time low in 2024 as law enforcement crackdowns, stronger defenses, and shifting policies make attacks riskier and less profitable.