ASGI Web Framework with Dependency Injection
Project description
Project Selva
Selva is a Python ASGI web framework built on top of asgikit and inspired by Spring Boot, AspNet Core and FastAPI.
Usage
Create an application and controller
from selva.web import Application, controller, get
@controller
class Controller:
@get
def hello(self):
return "Hello, World!"
app = Application(Controller)
Add a service
from selva.di import service
from selva.web import Application, controller, get
@service
class Greeter:
def greet(self, name: str) -> str:
return f"Hello, {name}!"
@controller
class Controller:
def __init__(self, greeter: Greeter):
self.greeter = greeter
@get
def hello(self):
return self.greeter.greet("World")
app = Application(Greeter, Controller)
Get parameters from path
from selva.di import service
from selva.web import Application, controller, get
@service
class Greeter:
def greet(self, name: str) -> str:
return f"Hello, {name}!"
@controller
class Controller:
def __init__(self, greeter: Greeter):
self.greeter = greeter
@get("hello/{name}")
def hello(self, name: str):
greeting = self.greeter.greet(name)
# A json response will be created from the returned dict
return {"greeting": greeting}
app = Application(Greeter, Controller)
Configurations with Pydantic
from selva.di import service
from selva.web import Application, RequestContext, controller, get
from pydantic import BaseSettings
class Settings(BaseSettings):
DEFAULT_NAME = "World"
@service
def settings_factory() -> Settings:
return Settings()
@service
class Greeter:
def __init__(self, settings: Settings):
self.default_name = settings.DEFAULT_NAME
def greet(self, name: str | None) -> str:
name = name or self.default_name
return f"Hello, {name}!"
@controller
class Controller:
def __init__(self, greeter: Greeter):
self.greeter = greeter
@get("hello/{name}")
def hello(self, name: str):
greeting = self.greeter.greet(name)
return {"greeting": greeting}
@get("hello")
def hello_optional(self, context: RequestContext):
name = context.query.get("name")
greeting = self.greeter.greet(name)
return {"greeting": greeting}
app = Application(Greeter, Controller)
Manage services lifecycle (e.g Databases)
from selva.di import service, initializer, finalizer
from selva.web import Application, RequestContext, controller, get
from pydantic import BaseSettings, PostgresDsn
from databases import Database
class Settings(BaseSettings):
DEFAULT_NAME = "World"
DATABASE_URL: PostgresDsn
@service
def settings_factory() -> Settings:
return Settings()
@service
class Repository:
def __init__(self, settings: Settings):
self.database = Database(settings.DATABASE_URL)
async def get_greeting(self, name: str) -> str:
result = await self.database.fetch_one(
query="select text from greeting where name = :name",
values={"name": name}
)
return result.text
@initializer
async def initialize(self):
await self.database.connect()
print("Database connection opened")
@finalizer
async def finalize(self):
await self.database.disconnect()
print("Database connection closed")
@service
class Greeter:
def __init__(self, repository: Repository, settings: Settings):
self.repository = repository
self.default_name = settings.DEFAULT_NAME
async def greet(self, name: str | None) -> str:
name = name or self.default_name
return await self.repository.get_greeting(name)
@controller
class Controller:
def __init__(self, greeter: Greeter):
self.greeter = greeter
@get("hello/{name}")
def hello(self, name: str):
greeting = self.greeter.greet(name)
return {"greeting": greeting}
@get("hello")
def hello_optional(self, context: RequestContext):
name = context.query.get("name")
greeting = self.greeter.greet(name)
return {"greeting": greeting}
app = Application(settings_factory, Repository, Greeter, Controller)
Define controllers and services in a separate module
├───application
│ ├───controllers.py
│ ├───repository.py
│ ├───services.py
│ └───settings.py
└───main.py
### application/settings.py
from selva.di import service
from pydantic import BaseSettings, PostgresDsn
class Settings(BaseSettings):
DEFAULT_NAME = "World"
DATABASE_URL: PostgresDsn
@service
def settings_factory() -> Settings:
return Settings()
### application/repository.py
from selva.di import service, initializer, finalizer
from databases import Database
from .settings import Settings
@service
class Repository:
def __init__(self, settings: Settings):
self.database = Database(settings.DATABASE_URL)
async def get_greeting(self, name: str) -> str:
result = await self.database.fetch_one(
query="select text from greeting where name = :name",
values={"name": name}
)
return result.text
@initializer
async def initialize(self):
await self.database.connect()
print("Database connection opened")
@finalizer
async def finalize(self):
await self.database.disconnect()
print("Database connection closed")
### application/services.py
from selva.di import service
from .settings import Settings
from .repository import Repository
@service
class Greeter:
def __init__(self, repository: Repository, settings: Settings):
self.repository = repository
self.default_name = settings.DEFAULT_NAME
async def greet(self, name: str | None) -> str:
name = name or self.default_name
return await self.repository.get_greeting(name)
### application/controllers.py
from selva.web import RequestContext, controller, get
from .services import Greeter
@controller
class Controller:
def __init__(self, greeter: Greeter):
self.greeter = greeter
@get("hello/{name}")
def hello(self, name: str):
greeting = self.greeter.greet(name)
return {"greeting": greeting}
@get("hello")
def hello_optional(self, context: RequestContext):
name = context.query.get("name")
greeting = self.greeter.greet(name)
return {"greeting": greeting}
### main.py
from selva.web import Application
# module named "application" is automatically registered
app = Application()
Websockets
from pathlib import Path
from selva.web import Application, FileResponse, RequestContext, controller, get, websocket
from selva.web.errors import WebSocketDisconnectError
@controller
class WebSocketController:
@get
def index(self) -> FileResponse:
return FileResponse(Path(__file__).parent / "index.html")
@websocket("/chat")
async def chat(self, context: RequestContext):
client = context.websocket
await client.accept()
print(f"[open] Client connected")
self.handler.clients.append(client)
while True:
try:
message = await client.receive()
print(f"[message] {message}")
await client.send_text(message)
except WebSocketDisconnectError:
print("[close] Client disconnected")
break
app = Application(WebSocketController)
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>WebSocket chat</title>
</head>
<body>
<form id="chat-form">
<textarea name="message-list" id="message-list" cols="30" rows="10" readonly></textarea>
<p>
<input type="text" name="message-box" id="message-box" />
<button type="submit">Send</button>
</p>
</form>
<script>
const messages = [];
const chat = document.getElementById("chat-form");
const textarea = document.getElementById("message-list");
textarea.value = "";
const messageInput = document.getElementById("message-box");
const socket = new WebSocket("ws://localhost:8000/chat");
function addMessage(message) {
messages.push(message)
textarea.value = `${messages.join("\n")}`;
textarea.scrollTop = textarea.scrollHeight;
}
chat.onsubmit = (event) => {
event.preventDefault();
const message = messageInput.value;
socket.send(message);
messageInput.value = "";
};
socket.onopen = (event) => {
console.log("[open] Client connected");
};
socket.onmessage = (event) => {
const message = event.data;
console.log(`[message] "${message}"`)
addMessage(message);
};
socket.onclose = (event) => {
if (event.wasClean) {
console.log(`[close] Connection closed cleanly, code=${event.code} reason=${event.reason}`);
} else {
console.log('[close] Connection died');
}
};
socket.onerror = function(error) {
console.log(`[error] ${error.message}`);
};
</script>
</body>
</html>
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
selva-0.1.0.tar.gz
(21.1 kB
view hashes)
Built Distribution
selva-0.1.0-py3-none-any.whl
(25.4 kB
view hashes)