Quick Reference

from apiup import Command, Context, command, Jsonable, Permission

# - method options: GET, POST, PUT, PATCH, DELETE
# - endpoint: 1. starts with /, 2. slugify using -, 3. to lower case
# - permission options: ADMIN, STAFF, VERIFIED_USER, USER, PUBLIC, OWNER, ONLY_OWNER
# Defaults:
# - method: POST
# - endpoint: /subapp/name-of-the-class
# - permission: STAFF
@command(method="GET", endpoint="/my-command", permission=Permission.OWNER)
class MyCommand(Command):
    async def command(self, ctx: Context) -> Jsonable:
        return {}  # dict, list, string, number, Document(s)

from apiup import Command, Context, command, Jsonable, F, Permission

@command()
class MyCommand(Command):
    field1: bool = F(type='bool', required=True, default=False)
    field2: int = F(type='int', choices=[1, 2, 3], min=0, max=10)  # min/max value
    field3: float = F(type='float', min=0, max=10)  # min/max value
    field4: str = F(type='string', format='email', choices=['a', 'b'], pattern='.*', min=0, max=10)  # min/max length
    field5: list[float] = F(type='list', subtype='float', min=0, max=10)  # min/max list size
    field6: dict[str, int] = F(type='dict', subtype='int', min=0, max=10)  # min/max dict size

    async def command(self, ctx: Context) -> Jsonable:
        print(self.field1, self.field2, ...)
        return self.get_data()

from apiup import command, Command, Context, Jsonable, Permission

@command()
class MyCommand(Command):
    async def command(self, ctx: Context) -> Jsonable:
        value: str = ctx.get_header('header-name')
        current_user = ctx.user
        current_user_id = ctx.user.get_id()
        return None

from apiup import Doc

Product = Doc('Product')
p = Product(name='...', price_cents=100)
p.name = '...'
p.price_cents = 100

# For usage (any module) -
from apiup import Doc
Product = Doc('Product')

# Basic data/model documents (any module)
from base_docs import ProductData, UserData

# For models with enhanced logic and services (app's products/docs.py)
from your_app_product.docs import Product

from apiup import Doc

User = Doc('User')
Product = Doc('Product')
doc: Product = Product()
doc.get_data()
doc.assign(assign_from_user=ctx.user)
await doc.save(by=ctx.user)
await doc.delete(by=ctx.user)
await doc.refresh(by=ctx.user)
await doc.push(by=ctx.user, field='some field', value='some value')
await doc.increment(by=ctx.user, field='some field', value=10)
await doc.load_lazy_fields()
product_user: User = await doc.ref('user_id')

from base_docs import ProductData

class Product(ProductData):
    def is_active():
        return self.status == 'active'

    async def set_as_active():
        self.status = 'active'
        await self.save()

    async def pre_create(self) -> None: pass
    async def pos_create(self) -> None: pass
    async def pre_update(self) -> None: pass
    async def pos_update(self) -> None: pass
    async def pre_delete(self) -> None: pass
    async def pos_delete(self) -> None: pass

# Usage:
from your_app.docs import Product

p = Product()
print(p.is_active())
await p.set_as_active()

from apiup import Q

# Searching a single Document
exists: bool = await Product.manager().exists(user=ctx.user, where=(Q.field1 == 'value1'))
doc: MyDoc | None = await MyDoc.manager().find_one(user=ctx.user, where=Q.field1 != 'value2')
doc: MyDoc | None = await MyDoc.manager().get_by_id('some id', user=ctx.user, cache_ttl=60)
doc: MyDoc | None = await MyDoc.manager().get_by(user=ctx.user, field1='value1', field2='value2', cache_ttl=60)

# It raises 404 Not Found error if no document was found
doc: MyDoc = await MyDoc.manager().get_by_id_or_raise('some id', user=ctx.user, cache_ttl=60)
doc: MyDoc = await MyDoc.manager().get_by_or_raise(user=ctx.user, field1='value1', field2='value2', cache_ttl=60)
doc: MyDoc = await MyDoc.manager().get_or_create(user=ctx.user, field1='value1', field2='value2', defaults=dict(field3='value3'))
doc: MyDoc = await MyDoc.manager().update_or_create(user=ctx.user, field1='value1', field2='value2', defaults=dict(field3='value3'))

# Searching multiple Documents - Cheat Sheet
docs: list[MyDoc] = await MyDoc.manager().all(user=ctx.user, limit=10, cursor_id=None)
docs: list[MyDoc] = await MyDoc.manager().find(user=ctx.user, where=Q.field1 >= 3, sort_by=['id'], limit=10, cursor_id=None)
docs: list[MyDoc] = await MyDoc.manager().text_search(
    'some word',
    user=ctx.user,
    fields=['name', 'username'],
    where=Q.active == True,
    sort_by=['id'],
    limit=10,
    cursor_id=None,
)
total: int = await MyDoc.manager().count(user=ctx.user, where=Q())

from base_docs import ProductData

class Product(ProductData):
    @classmethod
    async def find_active(cls) -> list['Product']:
        return await Product.manager().find_by(status='active')

# Usage:
print(await Product.find_active())

from base_docs import ProductData

class ProductServices:
    async def find_active(self) -> list['Product']:
        return await Product.manager().find_by(status='active')

class Product(ProductData):
    services: ProductServices = ProductServices()

# Usage:
print(await Product.services.find_active())

from apiup import Task

class ProductTask(Task):
    task_params: dict[str, Field] = dict(field=Field(type='string'))

    async def process(self, **kwargs):
        # Update the progress manually
        self.progress = 10  # 10%
        # Logic added here...

        return 'ok'

from apiup import Task

# Task progress calculated automatically
class ProductTask(Task):
    async def setup(self, **kwargs: typing.Any) -> None:
        pass

    async def get_items(self) -> list[typing.Any]:
        # Return list of objects to be processed
        return []

    # self.task_id available
    async def process_item(self, item) -> None:
        # process each item returned by get_items individually.
        pass

    async def finalize(self, task_report: Task.JSONType = None) -> Task.JSONType:
        # execute in the end of processing
        return None

    async def tear_down(self) -> None:
        # the last/finally execution (execute it even in case of exceptions)
        pass

# Trigger/Enqueue the ProductTask with the correspondent parameters to be executed in a Worker machine
await ProductTask.dispatch(param1=None, param2=None)

# Execute the task immediately
await ProductTask().execute_task(param1=None, param2=None)

from apiup import Q
Q()
Q.field
Q('field')
# Query comparison operators
Q.field == value
Q.field != value
Q.field > value
Q.field >= value
Q.field < value
Q.field <= value
# Query logical operators
(Q.field > value1) | (Q('field2') < value2)  # OR
(Q.field > value1) & (Q.field2 < value2)  # AND
Q.field._nor(Q())  # NOR
Q()._not()  # NOT
# Query search operators
Q.field._in(['value1', 'value2'])
Q.field.pattern('value1')
Q.field.ipattern('value1')
Q.field.iequal('value1')
Q.field.isearch('value1')
Q.list_field.contains('text')
Q.field.contains_text('text')
Q.field.contains_all([item1, item2])
Q.field.has_size(2)
Q.field.exists()