# Python

> Write custom APIs, hooks, schedulers and event listeners of API Maker in Python. main(g) as a function or a coroutine, g.req, g.res, g.sys.db, g.sys.system, g.sys.cache, the logger, errors, packages from PyPI, with many examples.

Source: https://docs.apimaker.dev/v4/docs/languages/python.html

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](https://docs.apimaker.dev/v4/docs/settings/sandboxSettings.html) 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**

```python
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 `dict`s 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**

```python
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**

```python
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**

```python
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**

```python
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**

```python
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**

```python
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**

```python
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**

```python
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**

```python
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**

```python
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**

```python
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**

```python
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**

```python
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**

```python
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**

```python
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**

```python
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**

```python
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](https://docs.apimaker.dev/v4/docs/utility-class/utility-class.html) written in Python is a module of its name : the other Python code of the account imports what it defines.

**The utility class Tax**

```python
from Round import cents      # another utility class

RATE = 0.2


def add(amount):
    return cents(amount * (1 + RATE))
```

**A custom API which uses it**

```python
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](https://docs.apimaker.dev/v4/docs/test-cases/test-cases.html#other-languages) 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**

```python
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](https://docs.apimaker.dev/v4/docs/settings/sandboxSettings.html#packages) (`requests` `2.32.3`), then import it :

**A package of PyPI**

```python
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](https://docs.apimaker.dev/v4/docs/settings/sandboxSettings.html#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](https://docs.apimaker.dev/v4/docs/languages/index.html#editor).

**Where the editor helps**

```python
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](https://docs.apimaker.dev/v4/docs/features/debugging.html).

## Related

- [Languages and runtimes](https://docs.apimaker.dev/v4/docs/languages/index.html) · [Sandbox settings](https://docs.apimaker.dev/v4/docs/settings/sandboxSettings.html) · [The global object g](https://docs.apimaker.dev/v4/docs/pre-defined-terms/global-object-g.html) · [g.sys.db examples](https://docs.apimaker.dev/v4/examples/sys/db/getAll.html)