ORM Basics#

# src/models/post.py
from asok import Model, Field, Relation

class Post(Model):
    title      = Field.String()
    body       = Field.String()
    slug       = Field.Slug(populate_from='title')
    published  = Field.Boolean(default=False)
    author_id  = Field.ForeignKey(lambda: User)
    cover      = Field.File(upload_to='posts')
    created_at = Field.CreatedAt()
    updated_at = Field.UpdatedAt()
    deleted_at = Field.SoftDelete()

class Category(Model):
    __tablename__ = "categories"  # Explicit table name
    name = Field.String()

asok make migration add_bio_to_users
asok migrate

class Product(Model):
    name = Field.text()
    price = Field.decimal()

    @staticmethod
    def scope_expensive(query, threshold=100):
        return query.where('price', '>', threshold)

# Usage:
products = Product.query().expensive(500).get()

class User(Model):
    email = Field.String(
        unique=True,      # Creates UNIQUE constraint + index
        nullable=False,   # NOT NULL constraint
        default="",       # Default value
        index=True,       # Creates database index for fast lookups
        hidden=True,      # Hide from forms/admin
        protected=True,   # Exclude from mass assignment
        label="Email Address",  # Custom form label
    )

class Project(Model):
    status = Field.Dropdown(
        choices=[("active", "Active"), ("archived", "Archived")],
        option__class="text-indigo-600 font-medium", # Target nested options
        menu__class="shadow-2xl rounded-xl border-none" # Target the menu
    )

    is_public = Field.Boolean(
        form_type="toggle",
        slider__class="bg-indigo-500", # Target the toggle slider
        container__class="p-4 bg-gray-50 rounded-lg"
    )

class Ticket(Model):
    status = Field.Dropdown(
        label="Ticket Status",
        choices=[("open", "Open"), ("pending", "Pending"), ("closed", "Closed")],
        searchable=True
    )

class Post(Model):
    category_id = Field.ForeignKey(
        Category, 
        dropdown=True,
        dropdown_title="name",      # Field to use as title
        dropdown_subtitle="desc",   # Optional subtitle field
        dropdown_image="icon_url",  # Optional image/avatar field
        dropdown_searchable=True
    )

class Contact(Model):
    name    = Field.String(max_length=100)  # short text, <input type="text">
    email   = Field.Email()                  # email input + auto-validation
    message = Field.Text()                   # long text, <textarea>

Contact.create(email='not-an-email', message='hi')
# → ModelError: Email is not a valid email address.

Field.String(max_length=80, default='draft', unique=True, nullable=False)

class Contact(Model):
    __tablename__ = "contacts"

    name = Field.String(
        max_length=100,
        nullable=False,
        label="Full Name",              # Custom label (default: "Name")
        rules="min:4|alpha_spaces",      # Custom validation rules
        messages={
            "required": "Please enter your full name",
            "min": "Name must be at least 4 characters",
            "max": "Name cannot exceed 100 characters",
            "alpha_spaces": "Name can only contain letters and spaces"
        }
    )

    email = Field.Email(
        max_length=100,
        nullable=False,
        label="Email Address",
        messages={
            "required": "Email is required to contact you",
            "email": "Please provide a valid email address"
        }
    )

    message = Field.Text(
        nullable=False,
        label="Your Message",
        rules="min:10",
        messages={
            "required": "Please tell us what you want to say",
            "min": "Message must be at least 10 characters"
        }
    )

form = Form.from_model(Contact, request)

required|max:100|min:4|alpha_spaces

pip install "asok[postgres-binary,redis]"

# Create
post = Post.create(title='Hello', body='...')

# Read
Post.find(id=1)
Post.find(slug='hello')
Post.all()
Post.all(published=True, order_by='-created_at', limit=10)
Post.count(published=True)
Post.exists(slug='hello')

# Update
post.title = 'Updated'
post.save()

# Or set + save in one call (perfect with form.data)
post.update(title='Updated', body='...')

# Delete
post.delete()

Post.first_or_create(slug='hello', defaults={'title': 'Hello', 'body': '...'})
Post.update_or_create(slug='hello', defaults={'title': 'Updated'})

Post.query() \
    .where('views', '>', 100) \
    .where('published', True) \
    .order_by('-created_at') \
    .limit(20) \
    .get()

Post.where('title', 'LIKE', '%python%').get()
Post.where_in('id', [1, 2, 3]).get()
Post.like('title', '%python%').get() # Shortcut for LIKE

query = Product.query().where('price', '>', 100)
print(query.to_sql())
# → SELECT * FROM products WHERE price > ?

print(query.raw_sql())
# → SELECT * FROM products WHERE price > 100

contacts = Contact.all()
print(contacts.to_sql())
# → SELECT * FROM contacts

# Useful for debugging complex relationship filters
user = User.find(id=1)
user_posts = user.posts  # This returns a ModelList
print(user_posts.raw_sql())
# → SELECT * FROM posts WHERE user_id = 1

from asok.orm import convert_sql_to_text
print(convert_sql_to_text(Product.query().limit(5)))

Post.raw("SELECT * FROM posts WHERE views > ?", [100])

class User(Model):
    name = Field.String()
    posts = Relation.HasMany('Post')
    profile = Relation.HasOne('Profile')
    roles = Relation.BelongsToMany('Role')

class Post(Model):
    author_id = Field.ForeignKey(lambda: User)
    author = Relation.BelongsTo('User')

user = User.find(id=1)
user.posts       # list of Post (property, no parentheses)
user.profile     # Profile or None
user.roles       # list of Role
post.author      # User or None

Relation.HasMany('Post', foreign_key='writer_id')
Relation.BelongsToMany('Tag', pivot_table='post_tags', pivot_fk='post_id', pivot_other_fk='tag_id')

posts = Post.query().with_('author').limit(20).get()
for p in posts:
    p.author  # served from cache, no extra query

post.attach('tags', [1, 2, 3])     # add pivot rows
post.detach('tags', [2])           # remove specific
post.detach('tags')                # remove all
post.sync('tags', [4, 5])          # replace all with these ids

class Post(Model):
    title = Field.String()
    deleted_at = Field.SoftDelete()

post.delete()              # soft delete
post.restore()             # un-delete
post.force_delete()        # permanent

Post.with_trashed().get()  # include deleted
Post.only_trashed().get()  # only deleted

class Post(Model):
    cover = Field.File(upload_to='posts')

class Post(Model):
    def before_save(self): ...
    def after_save(self): ...
    def before_create(self): ...
    def after_create(self): ...
    def before_update(self): ...
    def after_update(self): ...
    def before_delete(self): ...
    def after_delete(self): ...

result = Post.paginate(page=2, per_page=10, order_by='-created_at', published=True)

result['items']         # list of Post
result['total']         # total count
result['pages']         # total pages
result['current_page']  # 2

class User(Model):
    email = Field.String(unique=True)
    password = Field.Password()

user = User.create(email='a@b.com', password='secret')
user.check_password('password', 'secret')  # True

class Customer(Model):
    name = Field.String()
    ssn = Field.EncryptedString()  # Automatically encrypted/decrypted

customer = Customer.create(name="John", ssn="123-45-6789")

# In memory, the attribute is plaintext:
print(customer.ssn)  # "123-45-6789"

# In the database, the value is securely encrypted:
# e.g., "gAAAAABmX..."

slug = Field.Slug(populate_from='title')
# "Hello World!" → "hello-world"

post.to_dict()
# {"id": 1, "title": "Hello", ...}

from asok import Schema, Field

class PostSchema(Schema):
    title = Field.String()
    slug  = Field.String()

PostSchema().dump(post)
PostSchema(many=True).dump(posts)