Skip to content
This page View Markdown Open in ChatGPT Open in Claude

Python

Code in Python defines main(g), as a function or as a coroutine. It runs on the official image of Python (python:3.13-slim unless the settings name another version), with only the packages you add from PyPI.

In short

  • Write main(g), as a function or a coroutine.
  • It runs on python:3.13-slim, unless the settings name another version.
  • Refuse a request with raise g.Errors(message, code=400).
hello.py
1
2
def main(g):
    return {'hello': g.req.body.name or 'world'}
Runtime CPython 3.12 or later, a version of python:<version>-slim
main def main(g) runs on a thread of its own; async def main(g) runs on the event loop of the runner
Calls of g.sys typed : named arguments, each with its type (get_all("mysql_8", "inventory", "customers", limit=10)), and what a call answers has its type. They answer at once in a function, are awaited in a coroutine
Answer what main returns; None lets the API go on
Values plain JSON : objects are dicts whose keys are also attributes (g.req.body.id), a date is a text
Packages PyPI : requests, pandas>=2,<3, extras like uvicorn[standard]
Printing print() goes to the console of the account

The request and the answer

g.req and g.res have a property for each key they are known to have (g.req.body, g.req.query, g.res.status_code), and are dictionaries of JSON whose keys are also attributes. A key which is not there is None, as it is undefined in JavaScript, so g.req.body.name or 'world' needs no check.

Read the request, shape the answer
1
2
3
4
5
6
7
def main(g):
    customer_id = g.req.params.id                 # /api/custom-api/<path>/orders/:id
    status = g.req.query.status or 'open'          # ?status=paid
    token = g.req.headers['x-am-authorization']    # headers are in small letters
    g.res.status_code = 201
    g.res.headers['X-Request-Kind'] = 'orders'
    return {'customer': customer_id, 'status': status, 'signed': token is not None}
  • Change g.req in a pre hook : the API sees the change. Only what you change goes back to the server.
  • g.res.output holds the answer of the API in a post hook.
  • g.req.auth has the opened tokens : authAMUser, authAMDB (the row of your users table)…

Bytes as the answer

A file, a picture, a PDF : return its bytes (bytes, bytearray, memoryview), or put them in g.res.output, and name their kind of content. The answer is those bytes as they are, whatever the kind of content, with the headers of the code. (A text under a kind of bytes, image/gif, is read as base64.)

A picture as the answer
1
2
3
4
5
6
7
8
import base64


def main(g):
    picture = base64.b64decode('R0lGODlhAQABAIAAAP///////yH5BAEKAAEALAAAAAABAAEAAAICTAEAOwA=')
    g.res.content_type = 'image/gif'
    g.res.headers['Content-Disposition'] = 'inline; filename="pixel.gif"'
    return picture

Calling API Maker : g.sys

The methods of g.sys.db, g.sys.system and g.sys.cache have their Python names (get_all) and take named arguments, each with its type : what a call is about comes first and needs no name (the instance, the database, the table), the rest is given by name. What a call answers has its type : rows are a list[Row] (a dict whose keys are attributes too), a row a Row or None, a count an int, what an API describes a class with a property for each key (changed.updated_rows_count). A name which no call has is a TypeError of Python : a key is never sent under a wrong name.

Database calls in a coroutine
1
2
3
4
5
6
7
8
from am import G


async def main(g: G):
    rows = await g.sys.db.get_all('mysql_8', 'inventory', 'customers', limit=10, sort='-customer_id')
    one = await g.sys.db.get_by_id('mysql_8', 'inventory', 'customers', 2)
    count = await g.sys.db.count('mysql_8', 'inventory', 'customers', find={'country': 'IN'})
    return {'first': rows[0] if rows else None, 'two': one, 'in_india': count}

The same in a function : calls answer at once. (from am.sync import G tells an editor of Python so : the same g, with calls which are not awaited.)

Database calls in a function
1
2
3
4
5
6
7
8
9
from am.sync import G


def main(g: G):
    table = ('mysql_8', 'inventory', 'customers')
    saved = g.sys.db.save_single_or_multiple(*table, {'first_name': 'Ada', 'last_name': 'Lovelace'})
    g.sys.db.update_by_id(*table, 2, update_data={'first_name': 'MALLORY'})
    g.sys.db.remove_by_id(*table, 101)
    return saved.row    # what a save answers : saved.row, saved.rows, saved.is_list

What a call takes is also a class of am, made the same way : it is given in place of the named arguments, kept in a variable, and it is what the calls of several take. A dict with the keys as the API takes them is taken too (what a request brought).

What a call takes, as a value
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
from am import G, GetAll, GetToken


async def main(g: G):
    customers = GetAll('mysql_8', 'inventory', 'customers', find={'country': 'IN'}, limit=10)
    rows = await g.sys.db.get_all(customers)
    same = await g.sys.db.get_all({'instance': 'mysql_8', 'database': 'inventory', 'collection': 'customers'})
    tokens = await g.sys.system.get_tokens([GetToken('ada', 'secret'), GetToken('bob', 'secret')])
    whole = await g.sys.db.full.get_all(customers, get_total_count=True)   # the whole answer : success, data, errors…
    return {'rows': len(rows), 'same': len(same), 'tokens': len(tokens), 'ok': whole.success}

Run calls at the same time with asyncio.gather :

Calls at the same time
1
2
3
4
5
6
7
8
import asyncio

async def main(g):
    orders, products = await asyncio.gather(
        g.sys.db.count('mongodb', 'shop', 'orders'),
        g.sys.db.count('mongodb', 'shop', 'products'),
    )
    return {'orders': orders, 'products': products}

Rows as they arrive

get_all_by_stream and query_by_stream take what is asked (a GetAll, a Query) and hand the rows to your function as the database sends them : thousands of rows never sit in memory at once. The function can be a coroutine too.

Sum a large table
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
from am import GetAll


def main(g):
    total = {'amount': 0, 'rows': 0}

    def take(row):
        total['amount'] += row.amount or 0
        total['rows'] += 1

    g.sys.db.get_all_by_stream(GetAll('mysql_8', 'inventory', 'orders'), take)
    return total

Aggregations, generated APIs, plain queries

Other database calls
1
2
3
4
5
async def main(g):
    by_status = await g.sys.db.aggregate('mongodb', 'shop', 'orders', aggregate_query=[{'$group': {'_id': '$status', 'n': {'$sum': 1}}}])
    logs = await g.sys.db.gen.get_all_gen('mysql_8', 'inventory', 'logs')
    plain = await g.sys.system.execute_query('postgresql', 'SELECT 1', database='crm')
    return {'by_status': by_status, 'logs': len(logs), 'plain': plain}

System APIs and cache

Encrypt, hash, events and cache keys
1
2
3
4
5
6
7
async def main(g):
    sealed = await g.sys.system.encrypt({'card': '4111'})
    digest = await g.sys.system.hash('my text')
    await g.sys.system.emit_event('order_created', {'id': 7}, ['notify_shop'])
    await g.sys.cache.set_key('last_order', '7', ttl_seconds=600)
    last = await g.sys.cache.get_key('last_order')
    return {'sealed': sealed, 'digest': digest, 'last': last}
Call another service from the sandbox
1
2
3
4
async def main(g):
    answer = await g.sys.system.call_external_api('https://api.example.com/rates', method='GET', timeout=5000,
                                                  headers={'accept': 'application/json'})
    return answer.body        # answer.status_code, answer.headers, answer.body

Errors

Raise Errors (from am import Errors, also g.Errors) to answer with your own errors and status :

Refuse a request
1
2
3
4
5
6
def main(g):
    if not g.req.body.quantity:
        raise g.Errors("'quantity' is missing.", code=400)
    if g.req.body.quantity > 100:
        raise g.Errors([{'code': 400, 'message': 'At most 100 at a time.'}, {'code': 400, 'message': 'Split the order.'}])
    return {'ok': True}

A call of g.sys which fails raises Errors with the errors of the server. Catch it to decide what to do :

Catch the errors of a call
1
2
3
4
5
6
7
8
9
from am import Errors

async def main(g):
    try:
        await g.sys.db.remove_by_id('mysql_8', 'inventory', 'customers', 404)
    except Errors as refused:
        g.logger.warn('not removed :', refused.message)      # refused.code, refused.errors[0].field…
        return {'removed': False, 'why': refused.message, 'code': refused.code}
    return {'removed': True}

Any other exception answers 500 with its type, message and the lines of your code (ValueError: quantity must be positive, KeyError: the key 'qty' is not there). Calling sys.exit() is refused in words : return the answer instead.

Logs, printing and shared values

Logs and shared values
1
2
3
4
5
6
7
def main(g):
    g.logger.log('order', 7, 'paid')             # one line in the logs of the answer : "order 7 paid"
    g.logger.error({'step': 'payment', 'ok': False})  # objects as indented JSON
    print('kept in the console of the account')
    g.shared['calls'] = g.shared.get('calls', 0) + 1  # lives as long as the runner : shared by every request it serves
    g.res.shared.started = True                   # travels with this request : pre hook → API → post hook
    return g.shared['calls']

Hooks

A pre hook which limits every query of a table to the rows of the caller (row level security) :

Pre hook : only my rows
1
2
3
4
5
def main(g):
    person = (g.req.auth or {}).get('authAMDB') or {}
    find = g.req.query.find or {}
    find['owner_id'] = person.get('id')
    g.req.query.find = find

A post hook which shapes the answer of the API :

Post hook : hide a column
1
2
3
4
5
def main(g):
    rows = g.res.output or []
    for row in rows:
        row.pop('password', None)
    g.res.output = rows

What a hook returns ends the run : the API answers it. Return nothing to let the request go on.

Schedulers and event listeners

Scheduler : remove old logs every night
1
2
3
4
5
6
import datetime

async def main(g):
    before = (datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=30)).isoformat()
    removed = await g.sys.db.remove_by_query('mongodb', 'shop', 'logs', find={'createdAt': {'$lt': before}})
    g.logger.log('removed', removed.deleted_rows_count)
Event listener : react to an order
1
2
3
async def main(g):
    order = g.req.event_data or {}
    await g.sys.cache.set_key(f"order:{order.get('id')}", 'received', ttl_seconds=3600)

Utility classes

A utility class written in Python is a module of its name : the other Python code of the account imports what it defines.

The utility class Tax
from Round import cents      # another utility class

RATE = 0.2


def add(amount):
    return cents(amount * (1 + RATE))
A custom API which uses it
from Tax import add


async def main(g):
    return {"total": add(g.req.body.amount)}
  • The name of the class is the name of the module : letters of a to z, digits and _, no word of Python, and one class of that name among the Python classes of the account (whatever its folder).
  • The module is run once per sandbox, when it is first imported; a class which is saved again is what the next request imports.
  • In the Git repository the class is src/utility_classes/<folder>---<name>/<name>.py, and the package am (what from am import G imports, with the types of every call) is in am/python. pyproject.toml lists those folders for Pyright / Pylance (extraPaths), pytest (pythonpath) and mypy (mypy_path); to run the code yourself, put them on PYTHONPATH.

Test cases

A test case written in Python has a function per test : every function whose name starts with test_, plain or async. What it raises is its failure.

A test case
async def test_the_cost_of_a_parcel(g):
    await g.sys.test.mock("db.getAll", when={"collection": "rates"}, returns=[{"per_kg": 2.5}])
    out = await g.sys.test.run_custom_api("shippingCost", body={"weight": 4})
    assert out == {"cost": 10}, out
    assert len(await g.sys.test.calls("db.getAll")) == 1
  • g.sys.test : run_custom_api(name, version=…, body=…, query=…, params=…, headers=…), run_custom_api_full(…) for the whole answer (data, status_code, headers, logs), mock(method, when=…, returns=…), mocks([Mock(…), …]), clear_mocks(method=None), calls(method=None).
  • The custom API under test can be written in any language. The mocks of the Mocks tab of the test case answer too.

Packages

Add the package in Sandbox Settings (requests 2.32.3), then import it :

A package of PyPI
1
2
3
4
5
import requests

def main(g):
    answer = requests.get('https://api.example.com/rates', timeout=5)
    return answer.json()
  • Versions follow PEP 440 : 2.32.3 is that version, >=2,<3 and ~=2.32 are constraints.
  • A package which needs a system library (psycopg needs libpq5) gets it in the Dockerfile, above {{AM_PACKAGES}}.

How it runs

  • One runner process serves many requests at once : coroutines share the event loop, functions run on threads (64 at most per runner). CPU heavy code is better in a function; the sandbox opens more runners while all are busy.
  • The module of your code is compiled once and kept while it does not change : variables at module level live as long as the runner.
  • Values are plain JSON both ways : datetime becomes ISO text, Decimal a number, set and tuple lists, bytes base64, NaN and infinities null. What JSON can not hold (an open file) is an error which names its type.

In the code editor

The editor completes g as Python writes it, the names of your instances, databases and tables, and the columns of a table on the rows a call answers. More in the code editor in every language.

Where the editor helps
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from am import G


async def main(g: G):
    # g.  →  req, res, sys, logger, shared, Errors      g.sys.db.  →  get_all, get_by_id, query, gen, full…
    g.logger.log(g.req.body.name)  # g.req.  →  body, query, headers, auth…      g.res.  →  status_code, headers…
    rows = await g.sys.db.get_all('mysql_8', 'inventory', 'customers', limit=10)
    #      get_all('  →  your instances, then its databases, then their tables      , l  →  limit=, find=, sort=…
    for row in rows:
        g.logger.log(row.first_name, row['last_name'])  # row.  and  row['  →  the columns of customers
    return [row.first_name for row in rows]
  • Inside the brackets of a call the editor offers its named arguments, each with its type and what it is for (limit=, find=, select=…) ; inside find={' it offers the columns of the table, inside select=' and sort=' too. The same for the classes of what a call takes : GetAll(.
  • What a call answers has its type : the columns are offered after rows[0]., row., row[' and row.get(', in a for loop and in a list made from the rows ; changed. (an update_many) offers updated_rows_count, saved. offers row, rows and is_list, the whole answer of g.sys.db.full offers success, data, errors…
  • The table is the one the call names : what it is given first (get_all('mysql_8', 'inventory', 'customers')), named arguments (get_by_id(instance='mysql_8', ...)), a GetAll(…) kept in a variable, or a dict as the API takes it.
  • g.req. and g.res. offer a property for each key of the request and of the answer, g.req.auth. the opened tokens.

Reference

g. Python
g.req am.Request : body, query, params, headers, auth, req_info, event_data, is_api_request_from_user ; a dict with attribute access (am.Obj) too
g.res am.Response : status_code, content_type, headers, shared, output, errors, warnings ; an am.Obj too
g.sys.db get_all, get_by_id, query, save_single_or_multiple, master_save, array_operations, update_by_id, update_many, replace_by_id, remove_by_id, remove_by_query, aggregate, count, distinct, distinct_query, get_all_by_stream, query_by_stream
g.sys.db.gen the same with _gen : get_all_gen…
g.sys.db.full, g.sys.db.gen.full, g.sys.system.full, g.sys.cache.full the same calls, each with its whole answer : ApiResponse (success, status_code, data, total_count, errors, warnings)
g.sys.system encrypt, encrypt_with, decrypt, decrypt_with, hash, get_secret, get_secrets, call_external_api, call_external_apis, get_token, get_tokens, execute_query, get_table_meta, get_table_metas, create_indexes, drop_indexes, drop_indexes_of_many, get_indexes, get_indexes_of_many, emit_event, emit_event_ws, is_valid_data_for_table(s), is_valid_data_for_custom_api(s), multi_tenant_instance(s)_updated, is_valid_connection_string(s), get_lock, release_lock
g.sys.cache get_key, get_keys, set_key, set_keys, remove_key, remove_keys, reset_cache_db, reset_cache_dbs, reset_cache_custom_apis, reset_cache_custom_apis_of_tenant, reset_cache_system_apis, reset_cache_system_apis_of_tenant
g.sys.call('db.getAll', …) any API by its name in JavaScript, with its arguments as they are
what a call takes am.GetAll, am.GetById, am.Save, am.Query, am.DeepItem, am.GetToken, am.SetKey, am.Mock… : made with the same named arguments
g.logger log, debug, info, warn, warning, error
g.shared dict of the runner
am.Errors, g.Errors Errors(message, code=500) or Errors([{...}]); .code, .message, .errors (the list)

Debugging

  • The debug runner of Python runs debugpy, which the server alone reaches : the Debug tab of the API testing page shows where the code stopped.
  • A conditional breakpoint takes an expression of Python (len(items) > 3), a logpoint {expression}s of Python. The steps go past the runner and the packages (just my code).
  • Learn more Debugging.