Runtime API

PyGo Runtime API — Native Packages

PyGo Framework includes several native packages written in Go with Python-compatible APIs. These packages are bundled with the framework — no external dependencies required.

📦 Package List

Package	Category	Description
core	Core	AutoID, Enum, Array[T], Map[K]V, Optional, UUID, Email, DateTime
http	Core	Server, routing, middleware, templates
storage	Infrastructure	File upload, local + S3 compatible storage
cache	Infrastructure	Memory + Redis caching with TTL
queue	Infrastructure	Background jobs, memory + Redis backend
logger	Infrastructure	Structured JSON/text logging, 5 levels
validator	Functionality	Input validation, struct rules
mailer	Functionality	SMTP email, templates, bulk send
backup	Infrastructure	Database export, file archive, prune
multitenancy	Functionality	Tenant isolation, detection middleware
auth	Functionality	Sessions, JWT, OAuth, RBAC

---

Core

model

Base class for all data models.

model User:
  id: UUID
  email: Email
  name: String
  created: DateTime

AutoID

Auto-incrementing integer field.

model Post:
  id: AutoID
  title: String

Enum

Enumerated string type.

enum Status:
  draft
  published
  archived

model Article:
  status: Status

Array[T]

Array of any type.

model User:
  tags: Array[String]
  roles: Array[Enum]

Map[K]V

Map type.

model Config:
  settings: Map[String]String

Optional[T]

Nullable/Optional type.

model User:
  name: String
  phone: String?  # Optional

---

HTTP

Server

HTTP server built on Go’s net/http.

server = http.Server(
  host="0.0.0.0",
  port=8080,
)

route

Define HTTP routes.

handler hello:
  hello(name: String) -> String:
    return f"Hello, {name}!"

route GET /hello/:name -> hello

Middleware

HTTP middleware.

from http import Middleware

log_mw = Middleware(next => {
  log.info("Request", {"path": request.path})
  next()
})

server.use(log_mw)

---

Storage

LocalStore

Local filesystem storage.

from storage import LocalStore

store = LocalStore(
  basePath="./uploads",
  baseURL="/files",
)

# In handler
result = store.save(file)
return {"id": result["id"], "url": result["url"]}

S3Store

S3-compatible storage (requires configuration).

from storage import S3Store

store = S3Store(
  bucket="my-bucket",
  region="us-east-1",
  access_key=os.environ["AWS_ACCESS_KEY"],
  secret_key=os.environ["AWS_SECRET_KEY"],
)

---

Cache

NewMemoryCache

In-memory cache (auto-eviction by TTL).

from cache import NewMemoryCache

cache = NewMemoryCache()

cache.set("key", "value", ttl=3600)  # 1 hour
val = cache.get("key")

NewRedisCache

Redis-backed cache.

from cache import NewRedisCache

cache = NewRedisCache("redis://localhost:6379")

---

Queue

NewQueue

Job queue with auto-backend selection.

from queue import NewQueue

jobs = NewQueue()

# Dispatch a job
jobs.dispatch("send_email", {"to": "user@example.com", "subject": "Hello"})

Job Handlers (Python)

from queue import job_handler

@job_handler("send_email")
def send_email(payload):
    # This runs in Python
    from mailer import Send
    Send(to=[payload["to"]], subject=payload["subject"], html=payload["html"])

---

Logger

Default

Default logger instance.

from logger import Default

log = Default(
  format="json",
  level="info",
  service="myapp",
)

log.info("User logged in", {"user_id": user.id})
log.error("Database error", {"error": str(e)})
log.debug("Debug info", {"data": variable})

Levels

Level	Usage
debug	Detailed debug information
info	Key business events
warn	Expected errors (404, rate limit)
error	Unexpected errors
fatal	Critical errors (stops app)

---

Validator

ValidateStruct

Validate struct fields.

from validator import ValidateStruct

errors = ValidateStruct(model_instance)
if errors:
  log.error("Validation failed", {"errors": errors})

AddRule

Register custom validation rules.

from validator import AddRule

AddRule("strong_password", lambda p: len(p) >= 8)

---

Mailer

Send

Send a single email.

from mailer import Send

Send(
  to=["user@example.com"],
  subject="Welcome!",
  html="<h1>Welcome</h1>",
  text="Welcome!",
)

SendBulk

Send multiple emails.

from mailer import SendBulk

SendBulk(messages=[...])

WithTemplate

Load HTML email template.

from mailer import WithTemplate

tmpl = WithTemplate("emails/welcome.html")
html = tmpl.render({"name": user.name})

---

Backup

Manager

Backup manager for database and files.

from backup import Manager

backup = Manager()
backup.BackupDatabase("daily-backup")
backup.BackupFiles("./uploads", "uploads-backup")
backup.Prune(max_backups=7)

---

Multi-Tenancy

Store

Tenant registry and provider.

from multitenancy import Store

tenants = Store()
tenants.Register({"id": "acme", "name": "Acme Corp", "plan": "pro"})

Middleware

Tenant detection middleware.

from multitenancy import Middleware

server.use(Middleware(tenants))

---

Auth

Manager

Authentication manager.

from auth import Manager

auth = Manager(jwt_secret="secret", jwt_duration=24*time.hour)

generate_token

Create JWT token.

token = auth.generate_token({"uid": user.id, "role": user.role})

require_auth

Middleware requiring authentication.

route GET /dashboard -> auth.require_auth -> dashboard

require_role

Middleware requiring specific role.

route GET /admin -> auth.require_role("admin") -> admin_panel