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.
Installation
pip install selva
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(Controller, Greeter)
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(Controller, Greeter)
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(Controller, Greeter, settings_factory)
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(Controller, Greeter, Repository, settings_factory)
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.10.tar.gz
(21.0 kB
view hashes)
Built Distribution
selva-0.1.10-py3-none-any.whl
(25.4 kB
view hashes)