This is API example.
# An simple CRUD example.
import vetariasn as vt
class Employee(vt.orm.Base):
# vt.orm.Base is an subclass of sqlalchemy.orm.DeclarativeBase
# __tablename__ is optional. Convention over Configuration.
uid = vt.orm.op.Column(int, primary_key=True) # vt.orm.op == sqlalchemy
name = vt.orm.op.Column(str, unique=True, nullable=False)
class Message(vt.transient.Base):
# vt.transient: in-memory sqlite ORM API.
seq = vt.orm.op.Column(int, primary_key=True)
sender = vt.orm.op.Column(int, primary_key=True)
target = vt.orm.op.Column(int, primary_key=True)
content = vt.orm.op.Column(vt.orm.op.Text(), nullable=False)
# Create.
@vt.http.get("/create-employee")
async def create_employee(name: str):
async with vt.orm.Session() as session:
uid = vt.algo.calc_seqid() # Snowflake UID
# An simple creation operation
result = (await session.execute(
vt.orm.op.select(Employee)
.where(Employee.name == name)
)).fetchone()
if result is not None:
return {"uid": result._tuple()[0].uid, "already": True}
session.add(Employee(uid=uid, name=name))
await session.commit()
return {"uid": uid, "already": False}
# Read.
@vt.http.get("/get-employee/{name}")
async def get_employee_by_name(name: str):
async with vt.orm.Session() as session:
result = (await session.execute(
vt.orm.op.select(Employee)
.where(Employee.name == name)
)).fetchone()
if result is None:
return {"found": False, "uid": 0, "name": ""}
employee = result._tuple()[0]
return {"found": True, "uid": employee.uid, "name": employee.name}
# Start system.
if __name__ == "__main__":
vt.run()| Variable | Description |
|---|---|
| VETA_DB_URL | Database URL (Sqlalchemy Async) |
| VETA_HTTP_PORT | HTTP Web Port |