CatBase

CatBase

1. Introduction

1.1 Overview

CatBase is a statically typed programming language designed to be concise, easy to learn, and powerful. It supports common features such as network programming, multithreading, and file operations, and can import C language libraries.

The main features of CatBase include:

  • Concise and clear syntax
  • Static type system requiring explicit type declarations
  • Rich built-in function library
  • Supports network programming (TCP/UDP/HTTP)
  • Supports serial communication (RS-232/USB to serial)
  • Supports multithreaded programming
  • Compiles to native executables with excellent performance
  • Supports C library imports (.so/.a files)

1.2 CatBase Development Background

Why develop CatBase?

In the existing programming language ecosystem, we found several pain points:

Drawbacks of C:

  • Complex syntax with a steep learning curve
  • Requires manual memory management, prone to memory leaks
  • No built-in string type; requires character arrays
  • Lacks modern language features such as garbage collection and closures
  • Relatively low development efficiency

Advantages and drawbacks of Python:

  • Advantages: Concise and elegant syntax, high development efficiency, rich ecosystem
  • Drawbacks: Slow execution speed, cannot be directly compiled to native executables (requires the Python interpreter)
  • Drawbacks: Uses indentation to define code blocks, an error-prone approach (inconsistent indentation causes syntax errors)
  • Drawbacks: Variable declarations do not enforce type specification, easily causing type errors and increasing the time cost for programmers to identify variable types when reading code

CatBase's solution:

CatBase aims to combine the strengths of C and Python while solving Python's two problems:

Feature C Python CatBase
Execution efficiency High Low High
Development efficiency Low High High
Concise syntax No Yes Yes
Type safety Yes No Yes
Static typing Yes No Yes
Native execution Yes No Yes
Brace-delimited blocks Yes No Yes
Explicit type declarations Yes No Yes

Two Core Advantages of CatBase over Python

Advantage 1: Using braces to define code blocks

Python uses indentation to define code blocks, which easily causes syntax errors due to inconsistent indentation:


# Python - indentation issue example
def main():
    if True:
        print("Hello")  # Missing one space of indentation may cause a syntax error
    else:
        print("World")

CatBase uses braces {} to define code blocks, avoiding indentation issues:


# CatBase - brace-delimited code blocks
def main(args:list[str]) {
    if True {
        print("Hello")  # Braces clearly define the scope of the block
    } else {
        print("World")
    }
}

Advantage 2: Enforced variable type declarations

Python variables can omit type declarations; while concise, this is error-prone:


# Python - variable types are unclear
name = "Tom"          # string
age = 25              # integer
items = [1, 2, 3]    # list
# Programmers must read the code to infer variable types

CatBase requires explicit variable type declarations, making the code clear at a glance:


# CatBase - explicit variable types
name:str = "Tom"           # string type
age:int = 25               # integer type
items:list[int] = [1, 2, 3]  # list of integers type
# Variable types are clear at a glance, improving code readability

Why the name CatBase?

  • "Cat" represents simplicity and elegance (like a cat)
  • "Base" represents foundation and root, symbolizing a language oriented toward fundamental programming
  • At the same time, "CatBase" also implies that this language can serve as a foundation for learning more complex languages
  • Additionally, "Cat" happens to be a clever play on words as "C" at Base, which aligns with the language's goal

Code Volume Comparison

Let's compare the code volume of the three languages when implementing the same functionality:

Hello World:

C:


#include <stdio.h>
int main() {
    printf("Hello, World!\n");
    return 0;
}

Python:


print("Hello, World!")

CatBase:


def main(args:list[str]) {
    print("Hello, World!")
}

File Reading:

C:


#include <stdio.h>
#include <stdlib.h>

int main() {
    FILE *fp = fopen("test.txt", "r");
    if (fp == NULL) {
        printf("Cannot open file\n");
        return 1;
    }
    
    char buffer[1024];
    while (fgets(buffer, sizeof(buffer), fp) != NULL) {
        printf("%s", buffer);
    }
    
    fclose(fp);
    return 0;
}

Python:


with open("test.txt", "r") as f:
    print(f.read())

CatBase:


def main(args:list[str]) {
    f:File = file("test.txt", "r")
    content:str = f.read()
    f.close()
    print(content)
}

From the comparison, we can see that CatBase maintains Python-like conciseness while compiling to native executables with execution efficiency close to C.

1.3 Hello World

Below is the CatBase Hello World program:


def main(args:list[str]) {
    print("Hello, World!")
}

Output:


Hello, World!

1.4 Compiler and Command Line

The CatBase compiler (catcc) compiles .cat files into executables.

Basic Usage


# Compile source file
catcc source.cat

# Compile and run
catcc source.cat && ./source

# Or
catcc run source.cat

Command Line Parameters

Parameter Description
-v / --version Print the compiler version
-no-emit-obj Do not generate .o object file after compilation
-shared Generate a shared library (.so) instead of an executable
-static Static linking (the generated executable does not depend on dynamic libraries, suitable for minimal resource environments)
-O <level> Compilation optimization level: ReleaseFast, ReleaseSmall, ReleaseSafe
-o <output> Specify the output executable file name

Version Query

catcc -v / --version / -version directly prints the current compiler version, without depending on any source files. It does not read conf/config.conf, does not scan packages/, and is suitable for use in scripts / CI / documentation as a version detection command.


$ catcc -v
CatBase Compiler (catcc) 0.0.11

Usage Examples:


# 1. Manual version check
catcc -v
# → CatBase Compiler (catcc) 0.0.11

# 2. Version check in shell scripts
if [[ "$(catcc -v)" != *"0.0.11"* ]]; then
    echo "Requires catcc 0.0.11 or higher"
    exit 1
fi

# 3. CI minimum version assertion
REQUIRED="CatBase Compiler (catcc) 0.0.11"
if [[ "$(catcc --version)" != "$REQUIRED" ]]; then
    echo "CatBase version mismatch, requires $REQUIRED"
    exit 1
fi

# Do not generate .o file during compilation
catcc -no-emit-obj source.cat

# Generate shared library
catcc -shared mylib.cat

# Static linking (suitable for minimal resource environments, such as embedded systems)
catcc -static source.cat

# Generate the smallest executable
catcc -O ReleaseSmall source.cat

# Safety first (with runtime checks)
catcc -O ReleaseSafe source.cat

# Specify output file name
catcc -o myprogram source.cat

Optimization Level Description

The CatBase compiler provides three compilation optimization levels:

Optimization Level Description Applicable Scenarios
ReleaseFast Fast execution (default): fastest execution speed, larger code size Production environments, applications with high performance requirements
ReleaseSmall Smallest size: minimum code size, slightly slower execution Minimal resource environments (such as embedded devices, container image optimization), disk-space-constrained scenarios
ReleaseSafe Safety first: includes runtime checks, largest code size, slowest execution Development/debugging phases, scenarios requiring additional safety protection

Size Comparison Example:

Assume a simple Hello World program:

  • ReleaseFast: about 960 KB (default)
  • ReleaseSmall: about 13 KB (size reduced by about 98.6%)
  • ReleaseSafe: about 1.5 MB

Actual size depends on program complexity and the libraries used.

Minimal Size Example:

Using the -static -O ReleaseSmall combination can generate an executable with minimal size:


# Generate a minimal-size statically linked executable (about 13 KB)
catcc -static -O ReleaseSmall source.cat

This combination is especially suitable for:

  • Embedded systems
  • Container image optimization (such as Docker Alpine)
  • Resource-constrained Linux environments

Static Linking Description

The -static parameter makes the generated executable not depend on any dynamic link libraries (.so); all dependencies are statically linked into the executable.

Applicable scenarios:

  • Minimal resource Linux environments: such as embedded systems, container images (Alpine and other minimal distributions)
  • Simplified deployment: no need to install additional dynamic libraries on the target machine
  • High-security environments: reduces the attack surface of dynamic libraries

Notes:

  • Mutually exclusive with -shared: cannot use -static and -shared at the same time
  • Conflicts with .so/.a imports: if the code uses import to import .so or .a files, static linking will be automatically ignored with a warning

``bash Warning: -static is incompatible with .so import, ignoring static linking ``

  • C standard library: when statically linked, libc is linked into the executable

Difference between static linking and dynamic library import (.so):

File Type Description Can Be Statically Linked
.so (Shared Object) Dynamic link library, loaded at runtime ❌ No
.a (Archive) Static archive library, linked at compile time ✅ Yes

Why can't .so be statically linked?

.so files are runtime dependencies stored in the dynamic library directories of the target system (such as /usr/lib). Static linking happens at compile time and can only link code visible at compile time into the executable.

Example: MySQL Scenario

Assume your code imports the MySQL client library:


import "./libmysqlclient.so"

def main(args:list[str]) {
    # Connect to the database...
}

Compile with -static:


catcc -static source.cat

The compiler will issue a warning and ignore static linking:


Warning: -static is incompatible with .so import, ignoring static linking

This is because .so is a runtime dependency and cannot be packaged into the executable at compile time.

Solution:

  1. Use dynamic linking (default): the target machine needs the MySQL client library installed

``bash catcc source.cat ``

  1. Use a static archive library (.a): if a static library version is available

``bash catcc -static import "./libmysqlclient.a" source.cat `` Note: the static archive library must exist and be compatible with your target architecture

Example comparison:


# Dynamic linking (default)
catcc source.cat
# Generated source depends on: libc.so, libm.so, etc.

# Static linking
catcc -static source.cat
# Generated source does not depend on any dynamic libraries

# View the dynamic library dependencies of the executable
ldd ./source

# Statically linked executable
ldd ./source
# Output: "not a dynamic executable"

Compilation Output

After successful compilation, the following files are generated:

  • source - executable file (no extension)
  • source.o - object file (can be disabled with -no-emit-obj)
  • out/source.zig - generated intermediate code
  • libsource.so - shared library (generated when the -shared parameter is used)

Application Scenarios

Minimal resource Linux environments and server Linux environments, so when calling .so, both the server environment and the minimal compact environment need to be considered.

1.5 IDE Configuration

CatBase supports development in VS Code / Trae and other VS Code-based editors, providing syntax highlighting and code completion features.

Installation Steps

Method 1: Manual Installation

  1. Create a catbase folder in the extensions directory:
    • Windows: %USERPROFILE%\.vscode\extensions\catbase
    • Mac/Linux: ~/.vscode/extensions/catbase
  2. Create the following file structure in that folder:

catbase/
├── package.json
├── language-configuration.json
└── syntaxes/
    └── catbase.tmLanguage
  1. The contents of each file are as follows:

package.json (extension configuration)


{
  "name": "catbase-language",
  "displayName": "CatBase Language",
  "description": "CatBase programming language support for VS Code",
  "version": "1.0.0",
  "publisher": "catbase",
  "engines": {
    "vscode": "^1.60.0"
  },
  "categories": [
    "Programming Languages"
  ],
  "contributes": {
    "languages": [
      {
        "id": "catbase",
        "aliases": ["CatBase", "catbase"],
        "extensions": [".cat"],
        "configuration": "./language-configuration.json"
      }
    ],
    "grammars": [
      {
        "language": "catbase",
        "scopeName": "source.catbase",
        "path": "./syntaxes/catbase.tmLanguage"
      }
    ]
  }
}

language-configuration.json (language feature configuration)


{
  "comments": {
    "lineComment": "#",
    "blockComments": ["/*", "*/"]
  },
  "brackets": [
    ["{", "}"],
    ["[", "]"],
    ["(", ")"]
  ],
  "autoClosingPairs": [
    ["{", "}"],
    ["[", "]"],
    ["(", ")"],
    ["\"", "\""],
    ["'", "'"]
  ],
  "surroundingPairs": [
    ["{", "}"],
    ["[", "]"],
    ["(", ")"],
    ["\"", "\""],
    ["'", "'"]
  ],
  "indentationRules": {
    "increaseIndentPattern": "\\{[^}]*$",
    "decreaseIndentPattern": "^\\s*\\}"
  }
}

syntaxes/catbase.tmLanguage (syntax highlighting definition)


name: CatBase
scopeName: source.catbase
patterns:
  - include: '#comments'
  - include: '#strings'
  - include: '#keywords'
  - include: '#types'
  - include: '#numbers'
  - include: '#functions'

repository:
  comments:
    patterns:
      - name: comment.line.catbase
        begin: "#"
        end: "$"
      - name: comment.block.catbase
        begin: "/\\*"
        end: "\\*/"

  strings:
    patterns:
      - name: string.quoted.double.catbase
        begin: "\""
        end: "\""
        patterns:
          - name: string.escape.catbase
            match: \\.
      - name: string.quoted.single.catbase
        begin: "'"
        end: "'"
        patterns:
          - name: string.escape.catbase
            match: \\.

  keywords:
    patterns:
      - name: keyword.control.catbase
        match: \b(def|if|else|for|while|return|break|try|catch|except|finally|thread|async|await|import|from|as|not|and|or|in|is)\b
      - name: keyword.other.catbase
        match: \b(True|False|None)\b

  types:
    patterns:
      - name: storage.type.catbase
        match: \b(i64|i32|i16|i8|u64|u32|u16|u8|f64|f32|bool|str|list|dict|bytes|any)\b

  numbers:
    patterns:
      - name: constant.numeric.catbase
        match: \b\d+(\.\d+)?\b
      - name: constant.numeric.hex.catbase
        match: \b0x[0-9a-fA-F]+\b

  functions:
    patterns:
      - name: entity.name.function.catbase
        match: '[a-zA-Z_][a-zA-Z0-9_]*(?=\s*\()'
  1. Restart VS Code / Trae

Method 2: Install using VSIX file (recommended)

The project root directory already contains a vscode-extension folder, which can be used to generate a VSIX installation package for installation.

  1. Generate VSIX installation package (execute on a Linux server):

```bash # Enter the project directory cd /path/to/CatBase_Worksp

# Use Python to generate the VSIX file python3 -c " import zipfile import os import shutil import tempfile

tmpdir = tempfile.mkdtemp() ext_dir = os.path.join(tmpdir, 'extension') os.makedirs(ext_dir)

# Copy all files to the correct location src_dir = 'vscode-extension' for item in os.listdir(src_dir): src = os.path.join(src_dir, item) dst = os.path.join(ext_dir, item) if os.path.isfile(src): shutil.copy2(src, dst) elif os.path.isdir(src): shutil.copytree(src, dst)

# Create the vsix vsix_path = 'catbase-language-1.0.0.vsix' with zipfile.ZipFile(vsix_path, 'w') as vsix: for root, dirs, files in os.walk(ext_dir): for f in files: full_path = os.path.join(root, f) arc_path = 'extension/' + os.path.relpath(full_path, ext_dir) vsix.write(full_path, arc_path)

shutil.rmtree(tmpdir) print('VSIX created: ' + vsix_path) " `` After execution, the catbase-language-1.0.0.vsix` file will be generated in the project root directory.

  1. Download the VSIX file to your local machine:

``powershell # Download from the remote server to local Windows scp username@serverIP:/path/to/CatBase_Worksp/catbase-language-1.0.0.vsix C:\Users\YourUsername\Downloads\ ``

  1. Install the VSIX:
    • Open VS Code / Trae
    • Press Ctrl+Shift+X to open the Extensions view
    • Click the ... menu in the upper-right corner
    • Select "Install from VSIX..."
    • Select the downloaded catbase-language-1.0.0.vsix file
    • After installation, press Ctrl+Shift+P, type Developer: Reload Window to reload

Method 3: Manually copy files

The project root directory already contains a vscode-extension folder; you can directly copy that folder to the extensions directory:


# Copy the entire folder to the extensions directory
cp -r vscode-extension ~/.vscode/extensions/catbase

# Windows can use:
# xcopy /E /I vscode-extension "%USERPROFILE%\.vscode\extensions\catbase"

Method 4: Install in development mode


cd vscode-extension
code --install-extension .

Features

After installation, .cat files will receive the following support:

  • Syntax highlighting (keywords, strings, numbers, types, etc.)
  • Comment highlighting (#, // line comments and /* */ block comments)
  • Bracket matching
  • Automatic indentation
  • File icon display

2. Basic Syntax

Chapter Overview: In the previous chapter, we learned about the basic concepts and development background of CatBase. This chapter will dive into the basic syntax of CatBase, including variable declarations, data types, comments, and more. Through this chapter, you will master the basic programming elements of CatBase, laying a solid foundation for subsequent learning of more complex features.

2.1 Variable Declaration

CatBase Design Advantage: Unlike Python's dynamic type system, CatBase adopts a static type system that requires all variables to be explicitly declared with a type. This design brings the following advantages:

  • Type safety: type errors can be caught at compile time, reducing runtime errors
  • Code readability: variable types are clear at a glance, facilitating team collaboration
  • Performance optimization: the compiler can perform more optimizations, improving execution efficiency

CatBase requires all variables to be explicitly declared with a type. Variable declarations use a colon : to separate the variable name from its type.

Syntax


variable_name:type = initial_value

Basic Type Example


def main(args:list[str]) {
    # integer type
    age:int = 25

    # float type
    price:float = 19.99

    # string type
    name:str = "CatBase"

    # boolean type
    is_active:bool = True

    # byte type (0-255 integer)
    ascii_a:byte = byte(65)         # 'A' ASCII code
    ascii_z:byte = byte(90)         # 'Z' ASCII code

    print("Name: ", name, "\n")
    print("Age: ", age, "\n")
    print("Price: ", price, "\n")
    print("Active: ", is_active, "\n")
    print("ASCII a: ", ascii_a, " (char: ", chr(ascii_a), ")\n")
    print("ASCII z: ", ascii_z, " (char: ", chr(ascii_z), ")\n")
}

Output:


Name: CatBase
Age: 25
Price: 19.99
Active: true
ASCII a: 65 (char: A)
ASCII z: 90 (char: Z)

byte Type

CatBase provides a byte type to represent an 8-bit unsigned integer (0-255), corresponding to Zig's u8. The byte type is ideal for:

  • ASCII character codes (e.g. 'A' = 65, '0' = 48)
  • Raw byte data (used together with bytes)
  • Memory-efficient integer storage (1 byte vs int's 8 bytes)

Key Features of byte:

  1. Automatic compatibility with int: byte and int can be assigned to each other without explicit conversion
  2. Constructor byte(x): truncates or wraps an int/float into the 0-255 range
  3. Accepts None: byte(None) = 0
  4. Convert to a single character via chr(): chr(byte(65)) = "A"
  5. Range limit: values outside 0-255 will be truncated/wrapped by Zig's @intCast
  6. Single-character bytes literal b"X": b"A" auto-promotes to byte(65) - no byte() constructor needed

def main(args:list[str]) {
    # byte constructor
    b1:byte = byte(65)        # 65
    b2:byte = byte(255)       # 255 (max)
    b3:byte = byte(0)         # 0 (min)
    b4:byte = byte(None)      # 0

    # Single-character bytes literal (Method C - auto-promotes to byte)
    b5:byte = b"A"            # 65 ('A')
    b6:byte = b"Z"            # 90 ('Z')
    b7:byte = b"\xFF"         # 255 (hex escape)
    b8:byte = b"\x00"         # 0 (null byte)

    # byte ↔ int automatic conversion
    n:int = 100
    b9:byte = n               # int → byte (automatic)
    n2:int = b9               # byte → int (automatic)

    # byte with chr
    c:str = chr(byte(65))     # "A"
    c2:str = chr(b5)          # "A" (literal also works)

    print("byte(65) = ", b1, "\n")
    print("b\"A\" = ", b5, "\n")
    print("b\"\\xFF\" = ", b7, "\n")
    print("chr(byte(65)) = ", c, "\n")
}

byte vs int:

Featurebyteint
Zig typeu8 (8-bit)i64 (64-bit)
Range0 - 255-9.2×1018 ~ 9.2×1018
Memory1 byte8 bytes
chr compatiblechr(b) = "char"chr(65) = "A"
Use casesBytes, ASCII codes, raw dataGeneral integer arithmetic

Note: Currently str(byte) does not return a numeric string (different from str(int)). If you need the "numeric string" form, convert to int first: str(int(b)). See "Key Difference Between chr/ord and str/int" in Section 7.5 for details.

2.2 Built-in Name Protection

CatBase protects built-in function names and type names to prevent users from accidentally overriding them and causing runtime errors.

Built-in Function Name Protection

The following function names are built-in functions and cannot be used as names for user-defined functions:

print, len, range, open, file, close, sleep, input, str, int, float, bool, list, dict, hex, bin, oct, chr, ord, exec, pow, round, abs, max, min, sum, type, json_loads, json_dumps, sorted, reversed, enumerate, zip, map, filter, etc.

Built-in Type Name Protection

The following type names are built-in types and cannot be used as variable names:

int, str, float, bool, list, dict, File, Thread, Mutex, Queue, Response, WebSocket, TCPSocket, TCPClient, UDPSocket, Serial, RecordStream, PlayStream, Config, None, etc.

Example


# Error example - cannot use built-in function names
def file() -> int {  # Compile error!
    return 1
}

# Error example - cannot use built-in type names
def main(args:list[str]) {
    Thread:Thread = 1  # Compile error!
}

Compilation will report errors:


CatBase compilation errors:
  line 3: cannot use 'file' as function name: 'file' is a built-in function
  line 5: cannot use 'Thread' as variable name: 'Thread' is a built-in type

List Type


def main(args:list[str]) {
    # list type
    numbers:list[int] = [1, 2, 3, 4, 5]
    
    print("Numbers: ", numbers, "\n")
    print("First: ", numbers[0], "\n")
}

Output:


Numbers: [1, 2, 3, 4, 5]
First: 1

Dictionary Type


def main(args:list[str]) {
    # dictionary type
    person:dict[str, str] = {"name": "Tom", "age": "20"}
    
    print("Person: ", person, "\n")
    print("Name: ", person["name"], "\n")
}

Output:


Person: {"name": "Tom", "age": 20}
Name: Tom

Custom Types (Object Types)

In addition to basic data types, CatBase provides a series of custom types (also called object types). These types are built-in objects provided by the CatBase runtime library, used to implement various functionalities.

Common custom types:

Type Description Creation Method
File File object file(filename, mode)
Response HTTP response object http_post(), http_get()
WebSocket WebSocket client websocket(url, headers[optional])
Serial Serial port object serial(port, baud_rate)
TCPSocket TCP socket (client/server) tcpsocket()
TCPClient TCP client connection (from TCPSocket.accept) -
UDPSocket UDP socket udpsocket()
RecordStream Recording stream recordStream(rate, channels, chunk, format, device_name, callback)
PlayStream Playback stream playStream(rate, channels, format, device_name, callback)
list[dict[str,str]] Device list getInputDeviceList(), getOutputDeviceList()
Mutex Mutex lock mutex()
Thread Thread handle thread worker(args)
Queue Message queue queue(maxsize)
Config Configuration file object config(filename)

Usage Example:


def main(args:list[str]) {
    # TCP client
    sock:TCPSocket = tcpsocket()
    sock.connect("example.com", 80)

    # TCP server
    server:TCPSocket = tcpsocket()
    server.bind("0.0.0.0", 8080)
    server.listen(128)
    client:TCPClient = server.accept()

    # UDP
    udp:UDPSocket = udpsocket()
    udp.sendto("Hello", "127.0.0.1", 9999)

    # WebSocket client
    ws:WebSocket = websocket("ws://example.com/ws", None)

    # Serial port
    ser:Serial = serial("/dev/ttyUSB0", 115200)

    # Configuration file
    cfg:Config = config("app.conf")

    # File object
    f:File = file("test.txt", "r")

    # Audio stream
    stream:RecordStream = recordStream(rate=16000, channels=1)

    # Resources need to be closed after use
    client.close()
    ws.close()
    ser.close()
    f.close()
    stream.close()
}

Characteristics of custom types:

  1. Need to be created using built-in functions or constructors
  2. Usually need to manually release resources (call the close() method)
  3. Support specific method calls, such as stream.read(), ws.send(), client.recv(), etc.
  4. Type names are capitalized to distinguish them from basic data types

Queue (Thread-safe Message Queue)

Queue is a thread-safe message queue, aligned with Python's queue.Queue API.

Creation method:


q:Queue = queue(maxsize)  # maxsize of 0 means an unlimited queue

Supported types: Queue supports int, float, str, bytes, and bool types (compile-time dispatch; generates corresponding fromXxx code based on put argument type). The type of the first element inserted determines the type of all subsequent elements; elements with mismatched types will be silently discarded.

put/get Multi-type Dispatch (new in v0.0.8):

Input Type Generated Zig Code QueueItem Internal
str QueueItem.fromStr(arg) Str variant
bytes QueueItem.fromBytes(arg, allocator) Bytes variant (new)
int QueueItem.fromInt(arg) Int variant
float QueueItem.fromFloat(arg) Float variant
bool QueueItem.fromInt(@intFromBool(arg)) Int variant (0/1)

Multi-type Access on get:

Target Type Generated Zig Code Return Value
var x: str q.get().getStr() Str
var x: bytes q.get().getBytes() []u8 (new)
var x: int q.get().getInt() i64
var x: float q.get().getFloat() f64

Notes:

  • User code is completely unchanged: q.put(data, -1) auto-dispatches based on data type
  • bytes type is cloned and owned by the queue (avoids data being freed when thread exits)
  • getBytes() transfers ownership; the returned []u8 must be freed by the caller

Common methods:

Method Description Example
put(item, timeout_ms) Blocking put, timeout_ms=-1 means wait indefinitely q.put(msg, -1)
put_nowait(item) Non-blocking put, silently discarded when full q.put_nowait(msg)
get(timeout_ms) Get with timeout, returns zero value of the corresponding type on timeout; 0 means non-blocking, -1 means wait indefinitely q.get(1000)
get_nowait() Non-blocking get, returns zero value of the corresponding type when empty q.get_nowait()
task_done() Mark a task as completed q.task_done()
join() Wait for all tasks to complete and for all child threads to finish q.join()
empty() Check whether the queue is empty q.empty()
full() Check whether the queue is full q.full()
qsize() Get the number of elements in the queue q.qsize()
get_maxsize() Get the maximum capacity of the queue q.get_maxsize()

Usage Example (Producer-Consumer Pattern):


def producer(q:Queue) {
    i:int = 0
    while i < 5 {
        msg:str = str(i * 100)
        q.put(msg, -1)  # blocking wait to put
        print("Producer: put ", msg, "\n")
        i = i + 1
    }
}

def consumer(q:Queue) {
    i:int = 0
    while i < 5 {
        if !q.empty() {
            msg:str = q.get_nowait()
            print("Consumer: get_nowait ", msg, "\n")
            q.task_done()
        } else {
            print("Consumer: queue is empty, waiting...\n")
        }
        sleep(1)
        i = i + 1
    }
}

def main(args:list[str]) {
    q:Queue = queue(10)
    thread producer(q)
    thread consumer(q)
    q.join()  # wait for all tasks and threads to complete
    print("Test completed!\n")
}

Notes:

  • timeout_ms parameter of put(item, timeout_ms): -1 means wait indefinitely, 0 means non-blocking, a positive number means the number of milliseconds to wait
  • join() waits for all messages in the queue to be processed (unfinished_tasks == 0) and for all child threads started via thread to finish
  • When using put_nowait(), if the queue is full, messages will be silently discarded
  • Queue clones str/bytes type data when inserting elements to ensure the data is owned by the queue, preventing data from being released when the thread exits

Explicit Type Declaration Requirement

CatBase requires all variables to be explicitly declared with a type. The following code will cause an error:


def main(args:list[str]) {
    x = 10  # Error: must explicitly declare the type
}

Compile error:


line 3: [Type Error] Variable 'x' must be explicitly declared with a type

Correct way:


def main(args:list[str]) {
    x:int = 10  # Correct
}

Global Variables

CatBase supports declaring top-level global variables outside functions. Global variables retain their values throughout program execution and can be used to define constants, configuration parameters, etc.

Syntax:


variable_name:type = initial_value

Example:


# Global variable declaration (outside functions)
MY_NAME: str = "CatBase"
MY_VERSION: int = 1
MY_PI: float = 3.14
IS_ACTIVE: bool = true

def main(args:list[str]) {
    print("Name: ")
    print(MY_NAME)
    print("\n")
    print("Version: ")
    print(MY_VERSION)
    print("\n")
    print("PI: ")
    print(MY_PI)
    print("\n")
}

Output:


Name: CatBase
Version: 1
PI: 3.14

Characteristics of global variables:

  1. Global variables must be declared outside functions
  2. Global variables must be initialized at declaration
  3. Global variables can be accessed and modified in any function
  4. Global variables retain their values throughout program execution
  5. Global variables do not support the var keyword; type annotations must be used

2.2 Function Definition

Functions are defined using the def keyword.

Syntax


def function_name(parameter:type, ...) -> return_type {
    # function body
    return value
}

Note: The return type is optional. If no return type is specified, the function returns None by default.

Parameterless Function


def greet() {
    print("Hello, World!")
}

def main(args:list[str]) {
    greet()
}

Output:


Hello, World!

Function with Parameters


def greet(name:str) {
    print("Hello, ", name, "!")
}

def main(args:list[str]) {
    greet("CatBase")
    greet("World")
}

Output:


Hello, CatBase!
Hello, World!

Function with Return Value


def add(a:int, b:int) -> int {
    return a + b
}

def main(args:list[str]) {
    result:int = add(5, 3)
    print("5 + 3 = ", result, "\n")
}

Output:


5 + 3 = 8

Main Function

Every CatBase program must contain a main function as the entry point. The main function takes a list-type parameter to obtain command-line arguments.


def main(args:list[str]) {
    print("Program started!\n")
    print("Number of arguments: ", len(args), "\n")
}

Output:


Program started!
Number of arguments: 1
Getting Command-Line Arguments

Command-line arguments are passed in through the args list, where:

  • args[0] is the program name (executable file path)
  • args[1] onwards are the actual arguments passed by the user

def main(args:list[str]) {
    print("Program name: ", args[0], "\n")
    print("Number of arguments: ", len(args), "\n")
    
    # Iterate through all arguments
    i:int = 0
    while i < len(args) {
        print("args[", i, "] = ", args[i], "\n")
        i = i + 1
    }
}

Output:


Program name: ./hello
Number of arguments: 4
args[0] = ./hello
args[1] = hello
args[2] = world
args[3] = 123
Compile and Run Example

# Compile
catcc hello.cat

# Run with arguments
./hello hello world 123

2.2.4 Two Forms of main Function (Full vs Simplified)

CatBase provides two forms of the main function to suit different scenarios:

FormSignatureUse caseReceives command-line args
Full formdef main(args: list[str]) { }Programs that need to read user-provided command-line arguments✅ Yes
Simplified formdef main() { }Small tools / scripts that do not need command-line arguments❌ No (silently ignored)
1. Full Form: Explicitly Receive Command-Line Arguments

When you need to read user-provided arguments, you must use the full form:


def main(args: list[str]) {
    print("Number of args: ", len(args), "\n")
    if len(args) > 1 {
        print("First user arg: ", args[1], "\n")  # args[0] is the program name
    }
}

Run it:


./myprog hello world
# Output:
Number of args: 3
First user arg: hello
2. Simplified Form: Omit the Parameter Declaration

If you are sure your program does not need any command-line arguments, you can simply write def main() and omit the parameter declaration:


def main() {
    print("Hello from simplified main!\n")
    print("This program does not need command-line arguments.\n")
}

The compiler emits a soft warning (does not abort compilation):


CatBase compilation warnings (non-fatal):
  --> test_main_simplified.cat:25
  |
   > |   25 | def main() {
     |   26 |     print("Hello from simplified main!\n")
  |
  [main-simplified] main function declared with no parameters.
    Hint: command-line arguments passed to the compiled binary will be ignored at runtime.
    To receive command-line arguments, declare main as: def main(args: list[str]) { ... }

Behavior at runtime:

  • Compilation succeeds (warning does not abort)
  • The binary runs normally
  • User-provided command-line arguments are silently ignored (no error, no side effect)

./test_main_simplified             # OK, runs normally
./test_main_simplified a b c       # OK, a b c are ignored
./test_main_simplified --help      # OK, --help is ignored
3. Compiler Rules
  • The simplified form does not generate the code that reads command-line arguments (smaller binary, faster startup)
  • The two forms cannot coexist in the same program (each CatBase program allows only one main function)
  • The simplified form does not error, only warns; it does not leak any low-level language details
  • If the simplified main body still tries to read arguments (the compiler will not error), it cannot succeed — switch back to the full form in that case
4. Recommendation by Scenario
ScenarioRecommended form
CLI tool (parses user arguments)Full form def main(args: list[str])
GUI program / background service / daemonSimplified form def main()
Teaching example / Hello WorldSimplified form def main()
One-shot script / file-processing toolSimplified form def main()
Library / framework demo() functionSimplified form def main()
5. Side-by-side Comparison

# ===== Full form (receives arguments) =====
def main(args: list[str]) {
    print("Program: ", args[0], "\n")
    i: int = 1
    while i < len(args) {
        print("Arg ", i, ": ", args[i], "\n")
        i = i + 1
    }
}

# ===== Simplified form (no arguments) =====
def main() {
    print("This is a simple program with no command-line arguments.\n")
}

2.3 Comments

CatBase supports three styles of comments:

1. Hash Comment (#)

Use # for single-line comments; everything from # to the end of the line is a comment.


def main(args:list[str]) {
    # This is a comment
    print("Hello")  # This is also a comment
}

Output:


Hello

2. Block Comment (/* ... */)

Use /* and */ to wrap multi-line comment content.


def main(args:list[str]) {
    /*
     * This is a
     * multi-line comment
     */
    print("Hello")
}

Output:


Hello

3. Triple-Quote Comment (""" ... """)

Use three double quotes or single quotes to wrap multi-line comment content.


def main(args:list[str]) {
    """
    This is a
    triple-quote comment
    """
    print("Hello")
}

Output:


Hello

4. Double Slash Comment (//)

Use // for single-line comments; everything from // to the end of the line is a comment. This comment style is consistent with many mainstream programming languages (such as C++, Java, JavaScript).


def main(args:list[str]) {
    // This is a comment
    print("Hello")  // This is also a comment
}

Output:


Hello

2.4 Code Blocks

CatBase uses curly braces {} to define code blocks.


def main(args:list[str]) {
    # Simple code block
    {
        x:int = 10
        print("x = ", x, "\n")
    }
    
    # Code block in conditional statement
    if True {
        print("True block\n")
    } else {
        print("False block\n")
    }
}

Output:


x = 10
True block


3. Data Types

Chapter Overview: In the previous chapter, we learned the basic syntax of variable declarations. This chapter will dive into the data types supported by CatBase. Mastering these data types is the foundation for writing correct programs. CatBase provides a rich set of data types, including integers, floats, strings, booleans, lists, and dictionaries, to meet various programming needs.

3.1 Basic Data Types

CatBase supports the following basic data types:

Type Description Example
int Integer 42, -10
float Float 3.14, -0.5
str String "Hello"
bool Boolean True, False
list List [1, 2, 3]
dict Dictionary {"key": "value"}
bytes Byte sequence b"Hello"
any Universal type None, 42, "hi"
function Function reference my_callback

Integer Type


def main(args:list[str]) {
    a:int = 10
    b:int = -20
    c:int = 0
    
    print("a = ", a, "\n")
    print("b = ", b, "\n")
    print("c = ", c, "\n")
}

Output:


a = 10
b = -20
c = 0

Float Type


def main(args:list[str]) {
    pi:float = 3.14159
    neg:float = -2.5
    zero:float = 0.0
    
    print("pi = ", pi, "\n")
    print("neg = ", neg, "\n")
    print("zero = ", zero, "\n")
}

Output:


pi = 3.14159
neg = -2.5
zero = 0.0

Integer and Float Subtypes

In addition to the basic int and float types, CatBase supports more fine-grained numeric subtypes for precisely controlling memory usage and value ranges. These subtypes are particularly useful when calling external C library functions (via from...import declarations) and can exactly correspond to C language types.

Subtype Description Corresponding C/Zig Type Value Range
i8 8-bit signed integer i8 -128 ~ 127
i16 16-bit signed integer i16 -32768 ~ 32767
i32 32-bit signed integer i32 -2^31 ~ 2^31-1
i64 64-bit signed integer i64 -2^63 ~ 2^63-1
u8 8-bit unsigned integer u8 0 ~ 255
u16 16-bit unsigned integer u16 0 ~ 65535
u32 32-bit unsigned integer u32 0 ~ 2^32-1
u64 64-bit unsigned integer u64 0 ~ 2^64-1
f32 32-bit single-precision float f32 IEEE 754 single precision
f64 64-bit double-precision float f64 IEEE 754 double precision

Usage notes:

  • In general CatBase code, using int (corresponds to i64 by default) and float (corresponds to f64 by default) is sufficient
  • Subtypes are mainly used for parameter and return value types when declaring external functions with from...import, to ensure exact matching with C function types
  • int is equivalent to i64, and float is equivalent to f64

External function declaration example:


import "./libmylib.so" as mylib

# Use subtypes to declare external functions, exactly matching C function signatures
from mylib import process_byte(data: u8) -> i32
from mylib import get_coordinate() -> f32
from mylib import write_buffer(buf: bytes, len: u32) -> i64

def main(args:list[str]) {
    result:i32 = mylib.process_byte(65)
    print("Result: ", result, "\n")
}

String Type

CatBase supports using double quotes "..." or single quotes '...' to define strings, and also supports using f-string formatted strings:


def main(args:list[str]) {
    s1:str = "Hello"
    s2:str = 'World'
    s3:str = ""
    s4:str = ''
    
    print("s1 = ", s1, "\n")
    print("s2 = ", s2, "\n")
    print("s3 = '", s3, "'\n")
    print("s4 = '", s4, "'\n")
    print("s1 + s2 = ", s1 + " " + s2, "\n")
}

Output:


s1 = Hello
s2 = World
s3 = ''
s4 = ''
s1 + s2 = Hello World

f-string Formatted Strings

CatBase supports using f-strings (formatted strings) to embed variables and expressions in strings. An f-string begins with f" or f' and uses {variable_name} to insert variable values into the string.


def main(args:list[str]) {
    name:str = "CatBase"
    version:int = 1
    
    # Basic f-string usage
    msg:str = f"Hello, {name}!"
    print(msg, "\n")
    
    # Embed multiple variables
    info:str = f"{name} version {version}"
    print(info, "\n")
    
    # Embed expressions
    a:int = 10
    b:int = 20
    result:str = f"Sum = {a + b}"
    print(result, "\n")
}

Output:


Hello, CatBase!
CatBase version 1
Sum = 30

Syntax notes:

  • f"...{var}..." - embed variables in double-quoted strings
  • f'...{var}...' - embed variables in single-quoted strings
  • {expr} - the curly braces can contain any expression

Boolean Type


def main(args:list[str]) {
    t:bool = True
    f:bool = False
    
    print("t = ", t, "\n")
    print("f = ", f, "\n")
}

Output:


t = True
f = False

3.2 Composite Data Types

List

A list is an ordered, mutable collection.


def main(args:list[str]) {
    # Create a list
    nums:list[int] = [1, 2, 3, 4, 5]
    
    # Access elements
    print("First: ", nums[0], "\n")
    print("Last: ", nums[4], "\n")
    
    # Modify elements
    nums[0] = 10
    print("Modified: ", nums, "\n")
    
    # List length
    print("Length: ", len(nums), "\n")
}

Output:


First: 1
Last: 5
Modified: [10, 2, 3, 4, 5]
Length: 5

Dictionary

A dictionary is a key-value pair collection.


def main(args:list[str]) {
    # Create a dictionary
    person:dict[str, str] = {"name": "Tom", "age": "20", "city": "Beijing"}
    
    # Access values
    print("Name: ", person["name"], "\n")
    print("Age: ", person["age"], "\n")
    
    # Modify values
    person["age"] = 21
    print("Updated: ", person, "\n")
    
    # Add new key-value pairs
    person["country"] = "China"
    print("After add: ", person, "\n")
}

Output:


Name: Tom
Age: 20
Updated: {"name": "Tom", "age": 21}
After add: {"name": "Tom", "age": 21, "city": "Beijing", "country": "China"}

bytes

The bytes type represents a raw byte sequence, the core binary data type in CatBase. In the latest version, bytes has been implemented as a true []u8 slice (no longer wrapped in runtime.Str), and can be directly used as a buffer for C libraries.

Key Differences from the Old Version:

Feature Old Version (Str Wrapper) New Version (True []u8)
Internal Representation runtime.Str struct []u8 slice (ptr + len)
Memory Layout 26+ bytes metadata 16 bytes (ptr + len) + data
FFI Usage ❌ Would corrupt struct Pass directly to C
Smart Allocation ✅ Small data: static pool, large data: heap
Memory Isolation Complex ✅ Simple and clear

bytes literal syntax: Use the b"..." prefix to create a byte sequence.


def main(args:list[str]) {
    # Create ASCII bytes
    data:bytes = b"Hello"
    print(data)
    print("\n")

    # Create bytes containing hexadecimal data
    # \xHH represents a hexadecimal byte value
    hex_data:bytes = b"\x01\x02\x03\x04"
    print("Hex data length: ")
    print(len(hex_data))
    print("\n")
}

Output:


Hello
Hex data length: 4

bytes Smart Allocation (bytes_alloc)

CatBase provides the bytes_alloc(n) function for smart allocation:

  • Small data (≤ 4096 bytes): Uses static pool (64 pre-allocated 4096-byte buffers, equivalent to "stack-like" allocation)
  • Large data (> 4096 bytes): Uses heap allocation
  • Pool exhausted: Automatically falls back to heap

Advantages of smart allocation:

  • ✅ Zero cost for small allocations (no malloc)
  • ✅ Reduced heap fragmentation
  • ✅ Suitable for embedded and real-time systems
  • ✅ Smart deallocation (bytes_free auto-detects source)

def main(args:list[str]) {
    # Small data - uses static pool
    small:bytes = bytes_alloc(64)       # static pool
    print("small bytes: ", len(small), "\n")

    # Large data - uses heap
    large:bytes = bytes_alloc(100000)   # heap allocation

    # 1MB image data
    image:bytes = bytes_alloc(1920 * 1080 * 3)  # 1080p RGB

    # Must explicitly free
    bytes_free(small)
    bytes_free(large)
    bytes_free(image)
}

bytes Indexing and Access

bytes supports array-like indexing access and assignment, returning byte type (0-255):


def main(args:list[str]) {
    data:bytes = bytes_alloc(10)

    # Write data
    data[0] = byte(72)   # 'H'
    data[1] = byte(101)  # 'e'
    data[2] = byte(108)  # 'l'

    # Read data
    b:byte = data[0]   # 72
    print("data[0] = ", b, "\n")

    # Length
    print("length: ", bytes_len(data), "\n")

    bytes_free(data)
}

bytes ↔ str Bidirectional Conversion

bytes and str can be converted to each other. Conversion is by copy, not reference (to avoid dangling pointers):


def main(args:list[str]) {
    # str → bytes
    s:str = "Hello"
    b:bytes = bytes(s)         # copy data, bytes has independent memory
    print("b[0] = ", b[0], "\n")  # 72

    # bytes → str
    s2:str = str(b)             # copy data, str has independent memory
    print("s2 = ", s2, "\n")   # "Hello"

    # None conversion
    empty_b:bytes = bytes(None)
    empty_s:str = str(None)
    print("empty bytes len: ", len(empty_b), "\n")
}

Important: Conversion is by copy, not reference:

  • bytes(s) creates independent bytes data (freeing s doesn't affect bytes)
  • str(b) creates independent str data (freeing b doesn't affect str)

pointer_of Smart Dispatch (bytes Data Pointer)

The pointer_of function now supports smart dispatch:

  • For bytes variables: Returns data pointer (*u8), directly passable to C functions
  • For other type variables: Returns variable address (*i64, *f64, etc.)

This means users don't need to learn a new APIpointer_of(bytes_var) automatically returns the data pointer!


def main(args:list[str]) {
    # bytes: pointer_of returns data pointer
    data:bytes = bytes_alloc(100)
    data[0] = byte(65)   # 'A'

    p:Pointer = pointer_of(data)   # points to data[0], directly passable to C
    if !p.is_null() {
        v:byte = p.get(byte)
        print("p.get(byte) = ", v, "\n")

        p.set(byte, 90)            # write 'Z'
        print("data[0] after set: ", data[0], "\n")
    }

    # int: pointer_of returns variable address
    n:int = 42
    p2:Pointer = pointer_of(n)
    p2.set(int, 100)
    print("n after set: ", n, "\n")

    bytes_free(data)
}

Comparison: In the old version, pointer_of(bytes_var) would return the slice struct address (wrong!), causing complete FFI failure. Now it directly returns the data pointer, with zero-cost FFI integration.

bytes Escape Sequences

The bytes type supports the following escape sequences:

Escape Sequence Description
\n Newline character (0x0A)
\t Tab character (0x09)
\r Carriage return character (0x0D)
\" Double quote character
\\ Backslash character
\xHH Hexadecimal byte value (HH is a two-digit hexadecimal number)

bytes Literal Length-based Auto Type Promotion

CatBase's b"..." literal automatically determines its type based on the number of characters:

Literal Chars Auto Type Example
b"X" 1 byte (u8) b"A" → 65
b"AB" ≥ 2 bytes ([]u8) b"AB" → [65, 66]
b"" 0 bytes (empty) b"" → []

Design Rationale: A single-byte literal is conceptually equivalent to "one byte", so it auto-promotes to the byte type without needing the byte() constructor wrapper. This makes code more concise:


def main(args:list[str]) {
    # Single character → byte type
    b1:byte = b"A"             # 65
    b2:byte = b"\xFF"          # 255

    # Multi-character → bytes type
    bs1:bytes = b"Hello"        # [72, 101, 108, 108, 111]
    bs2:bytes = b"\x01\x02\x03"  # [1, 2, 3]

    # Use literals directly when assigning to bytes arrays
    buf:bytes = bytes_alloc(5)
    buf[0] = b"H"              # 72
    buf[1] = b"e"              # 101
    buf[2] = b"\x6C"           # 108 ('l')

    # Fully equivalent to the byte() constructor
    if b1 == byte(65) {
        print("✓ b\"A\" == byte(65)\n")
    }
}

Type Detection:


type(b"A")      # → "byte"
type(b"AB")     # → "bytes"
type(b"hello")  # → "bytes"
type(b"")       # → "bytes" (empty)

Notes:

  • ⚠️ b"A" must be a single character; more than 1 character is treated as bytes
  • ⚠️ b"" (empty) is bytes type, not byte
  • ✅ The byte(65) constructor is still preserved and continues to work

bytes Application Example: Serial Communication

The bytes type is well-suited for serial communication and can send binary protocol data:


def main(args:list[str]) {
    # Open serial port
    ser:Serial = serial("/dev/ttyUSB0", 115200)
    print("Serial port opened successfully\n")

    # Send text data
    ser.write(b"Hello, Serial!\n")

    # Send binary protocol data (Modbus example)
    # Frame format: device address(1) + function code(1) + data(N) + CRC(2)
    modbus_frame:bytes = b"\x01\x03\x00\x00\x00\x0A"
    ser.write(modbus_frame)

    # Read response
    data:str = ser.read(1024)
    print("Received: ")
    print(data)
    print("\n")

    # Close serial port
    ser.close()
    print("Serial port closed\n")
}

Output:


Serial port opened successfully
Serial port closed

bytes Application Example: Binary Protocol Construction


def main(args:list[str]) {
    # Construct the first 20 bytes of the IP header
    version_ihl:bytes = b"\x45"      # Version(4) + Header length(5)
    tos:bytes = b"\x00"              # Type of service
    total_length:bytes = b"\x00\x14"  # Total length (20)
    identification:bytes = b"\x00\x00" # Identification
    flags_offset:bytes = b"\x40\x00"  # Flags + Fragment offset
    ttl:bytes = b"\x40"              # Time to live (64)
    protocol:bytes = b"\x06"         # Protocol (TCP)
    checksum:bytes = b"\x00\x00"      # Checksum
    src_ip:bytes = b"\xC0\xA8\x01\x01" # Source IP (192.168.1.1)
    dst_ip:bytes = b"\xC0\xA8\x01\x02" # Destination IP (192.168.1.2)

    # Combine the complete IP header
    ip_header:bytes = version_ihl + tos + total_length + identification + flags_offset + ttl + protocol + checksum + src_ip + dst_ip

    print("IP header length: ")
    print(len(ip_header))
    print("\n")
    print("IP header hex: ")
    print(ip_header)
    print("\n")
}

Output:


IP header length: 20
IP header hex: E5 00 14 00 00 40 00 40 06 00 00 C0 A8 01 01 C0 A8 01 02

any Type (Universal Type)

any is CatBase's universal type that can store values of any type, including int, str, float, bool, list, dict, bytes, and None. It is implemented using a union type (Variant) internally, suitable for scenarios that require dynamically storing values of different types, such as:

  • Dictionary values after JSON parsing (dict[str, any])
  • Variables whose type is uncertain at runtime
  • Containers that need to store values of multiple types

Features:

  • any type variables can be assigned values of different types multiple times
  • The type() function returns an any:actual_type format for any type variables (e.g., any:int, any:str, any:None)
  • Type conversion functions like str(), int(), float() can be used to extract values of specific types from an any value
  • any type variables can be assigned None to represent a null value

Example:


# Test that any type can store values of various types
def main(args:list[str]) {
    # Create an any type variable, initialized to None
    x: any = None
    print("x (None) =", str(x), ", type =", type(x))

    # Store a string
    x = "hello"
    print("x (str) =", str(x), ", type =", type(x))

    # Store an integer
    x = 42
    print("x (int) =", str(x), ", type =", type(x))

    # Store a float
    x = 3.14
    print("x (float) =", str(x), ", type =", type(x))

    # Store a boolean
    x = True
    print("x (bool) =", str(x), ", type =", type(x))

    # Store a list
    x = [1, 2, 3]
    print("x (list) =", x, ", type =", type(x))

    # Store a dict
    x = {"key": "value"}
    print("x (dict) =", x, ", type =", type(x))

    print("\nAll types stored successfully in any variable!")
}

Output:


x (None) = None , type = any:None
x (str) = hello , type = any:str
x (int) = 42 , type = any:int
x (float) = 3.14 , type = any:float
x (bool) = true , type = any:bool
x (list) = [list] , type = any:list
x (dict) = [dict] , type = any:dict

All types stored successfully in any variable!

any Type and dict[str, any]:

The most common use of the any type is as the value type of dict[str, any] for handling JSON data. json_loads() returns the dict[str, any] type:


def main(args:list[str]) {
    # JSON parsing returns dict[str, any]
    data: dict[str, any] = {"name": "Alice", "age": 25, "score": 95.5}

    # Use any type to receive the return value of get
    name: any = data.get("name")
    age: any = data.get("age")
    missing: any = data.get("missing")   # Non-existent key returns None

    print("name =", str(name), ", type =", type(name))
    print("age =", str(age), ", type =", type(age))
    print("missing =", str(missing), ", type =", type(missing))

    # Extract specific types from any values
    name_str: str = str(data.get("name"))
    age_int: int = int(data.get("age"))
    score_float: float = float(data.get("score"))

    print("name_str =", name_str)
    print("age_int =", age_int)
    print("score_float =", score_float)
}

Output:


name = Alice , type = any:str
age = 25 , type = any:int
missing = None , type = any:None
name_str = Alice
age_int = 25
score_float = 95.5

Note:

  • After assigning a value to an any type variable, its declared type remains any and does not change to the actual value's type
  • Before performing arithmetic operations on any type variables, use type conversion functions (such as int(), float()) to extract the concrete value first
  • dict.get(key) returns None for dict[str, any] when the key is not found; for specific types like dict[str, int], it returns the zero value of that type (e.g., 0, "", false)

3.2.1 Type Guards and Narrowing (Recommended)

The most central pattern for using any variables is to first check the type with is, then use the variable as a concrete type inside the then block. The compiler automatically narrows the variable to that concrete type inside the then block; no manual conversion is required.

Supported is keywords:

KeywordMeaningExample
is intIs it an integer?if x is int { ... }
is floatIs it a float?if x is float { ... }
is strIs it a string?if x is str { ... }
is boolIs it a boolean?if x is bool { ... }
is listIs it a list?if x is list { ... }
is dictIs it a dict?if x is dict { ... }
is bytesIs it a byte sequence?if x is bytes { ... }
is NoneIs it None?if x is None { ... }
is not TIs it not type T?if x is not None { ... }

Basic Example:


def main(args: list[str]) {
    # Multiple branches to handle different types
    x: any = "hello"
    if x is int {
        print("integer:", x + 1, "\n")          # x is narrowed to int here
    } else if x is str {
        print("string:", str(len(x)), "\n")  # x is narrowed to str here
    } else if x is None {
        print("null value\n")
    } else {
        print("other type:", str(type(x)), "\n")
    }
}

is not reverse guard:


def main(args: list[str]) {
    # Use the value after excluding None
    x: any = 42
    if x is not None {
        if x is int {
            print("non-null int:", x * 2, "\n")
        }
    }
}

Chained guards (nested if):


def main(args: list[str]) {
    # Extract nested structures from dict[str, any] with multiple guards
    data: dict[str, any] = {"user": {"name": "Alice", "age": 30}}
    if data.get("user") is dict {
        user: any = data.get("user")
        if user.get("name") is str {
            name: any = user.get("name")
            print("user name:", str(name), "\n")
        }
    }
}

Type guards inside match statements:

match also supports type-name patterns (note: only type names are supported, not literal values):


def main(args: list[str]) {
    x: any = 42
    match x {
        int  => { print("int:", x + 1, "\n") }
        str  => { print("str:", x, "\n") }
        list => { print("list len:", len(x), "\n") }
        _      => { print("other type\n") }
    }
}

3.2.2 Explicit Type Conversion (Alternative When Guards Are Unavailable)

When you don't want to write a guard, or when an any value is already known to be a specific type, you can use the built-in conversion functions int(), str(), float(), bool() to extract the value directly.

str() — any → str:


def main(args: list[str]) {
    x: any = "world"
    s: str = str(x)     # Direct conversion, no guard needed
    print(s, "!\n")     # Output: world!
}

int() — any → int (note: first is int guard, otherwise it returns 0):


def main(args: list[str]) {
    # Method 1: assign directly after guard (recommended)
    x: any = 42
    if x is int {
        n: int = x       # Already confirmed int, assign directly
        print("n =", n, "\n")
    }

    # Method 2: without guard, int(any) returns 0 (default behavior)
    y: any = "100"
    m: int = int(y)      # y is not int, returns 0
    print("m =", m, "\n")
}

float() — any → float:


def main(args: list[str]) {
    x: any = 3.14
    if x is float {
        f: float = x
        print("f =", f, "\n")
    }
}

bool() — any → bool (requires an is guard first, because bool() on a Variant union takes the else branch):


def main(args: list[str]) {
    x: any = "yes"
    if x is str {
        s: str = x
        b: bool = bool(s)         # bool() works normally after the guard
        if b {
            print("non-empty string is treated as true\n")
        }
    }
}

str() conversion vs type guard — selection advice:

ScenarioRecommended Approach
any is already a known type but the compiler can't tellDirect conversion: str(x) / int(x)
any may be one of several types and requires type-branched handlingGuard: if x is T { ... }
Temporary variables used in operations (e.g. len, arithmetic, comparison)Use directly after if x is T guard

3.2.3 Container Nesting with any

The most common practical use of any is as the element type of a container, especially list[any] and dict[str, any].

dict[str, any] — the standard pattern for JSON / config parsing:


def main(args: list[str]) {
    # Simulated JSON data
    data: dict[str, any] = {
        "name": "Alice",
        "age": 30,
        "scores": [90, 85, 92],
        "active": True,
        "profile": None
    }

    # Basic field access + guard
    if data.get("name") is str {
        name: any = data.get("name")
        print("name:", str(name), "\n")
    }
    if data.get("age") is int {
        age: any = data.get("age")
        print("age:", age + 1, "\n")   # Can directly add 1 after guard
    }

    # Nested structure
    if data.get("scores") is list {
        scores: any = data.get("scores")
        print("scores len:", len(scores), "\n")
    }
}

list[any] — mixed-type list:


def main(args: list[str]) {
    items: list[any] = [1, "hello", True, 3.14, None]

    # Branch on type during iteration
    for item in items {
        if item is int {
            print("int:", item * 2, "\n")
        } else if item is str {
            print("str:", str(item), "\n")
        } else if item is None {
            print("None\n")
        }
    }
}

Multi-level nesting — dict[str, list[any]]:


def main(args: list[str]) {
    # Outer dict's value is a list, list elements are any
    matrix: dict[str, list[any]] = {
        "row1": [1, "two", 3.0],
        "row2": [True, None, "end"]
    }

    if matrix.get("row1") is list {
        row: any = matrix.get("row1")
        first: any = row[0]   # Index access
        if first is int {
            print("first int:", first, "\n")
        }
        second: any = row[1]
        if second is str {
            print("second str:", str(second), "\n")
        }
    }
}

Modifying any containers — operate after an is guard:


def main(args: list[str]) {
    # dict[str, any]: can call .put after guard; list: can call .append after guard
    d: dict[str, any] = {"a": 1}
    if d is dict {
        d.put("b", 2)             # .put can be called after guard
    }

    xs: list[any] = [1, 2]
    if xs is list {
        xs.append(3)              # .append can be called after guard
    }
    print("d =", d, "\n")
    print("xs =", xs, "\n")
}

3.2.4 any in Function Parameters and Return Values

Function returns any — caller must use an is guard:


def get_value(flag: int) -> any {
    if flag == 0 {
        return "text"
    } else if flag == 1 {
        return 42
    }
    return None
}

def main(args: list[str]) {
    v: any = get_value(0)
    if v is str {
        print("got str:", str(v), "\n")
    } else if v is int {
        print("got int:", v + 1, "\n")   # Use directly after guard
    } else {
        print("got None\n")
    }
}

Function parameter any — accepts any type:


def describe(x: any) -> str {
    if x is int {
        return "integer"
    } else if x is str {
        return "string"
    } else if x is list {
        return "list"
    }
    return "other"
}

def main(args: list[str]) {
    print(describe(42), "\n")        # integer
    print(describe("hi"), "\n")      # string
    print(describe([1, 2]), "\n")    # list
}

Note: A function parameter declared as x: any does not trigger the "cannot call method on any" error (because the parameter itself has no explicit any assignment semantics), but the function body still needs an is guard before calling methods or performing operations.

3.2.5 Cross-type Assignment and Round-trip

An any variable can be assigned values of different types repeatedly; type information is carried at runtime by the Variant union type:


def main(args: list[str]) {
    x: any = 100            # store int
    x = "now string"        # reassign to str
    x = [1, 2, 3]           # reassign to list
    x = {"k": "v"}          # reassign to dict
    print("final type:", str(type(x)), "\n")  # any:dict
}

typed ↔ any round-trip example:


def main(args: list[str]) {
    # typed → any → typed (information not lost)
    n: int = 42
    x: any = n              # int → any
    if x is int {
        m: int = x          # any → int
        print("round trip:", m, "\n")
    }

    # Container: dict[str, int] → any → take back and modify
    d: dict[str, int] = {"a": 1, "b": 2}
    x = d
    if x is dict {
        x.put("c", 3)        # can put after guard
        print("modified:", str(len(x)), "\n")  # 3
    }
}

3.2.6 any Cheat Sheet

NeedSyntaxNotes
Declare an any variablex: any = 42Explicitly declare the type as any
Store Nonex: any = NoneAny type can hold None
Check typeif x is T { ... }T = int / str / float / bool / list / dict / bytes / None
Reverse checkif x is not None { ... }Exclude a type
Use after guardif x is int { n := x; ... }x is narrowed to int in the then block
Convert explicitly to strs: str = str(x)No guard required; direct conversion
Convert explicitly to intn: int = int(x)Without a guard, non-int values return 0
Convert to floatf: float = float(x)Same as above
Get type explicitlytype(x)Returns the "any:actual_type" string
Container nestingdict[str, any]Standard pattern for JSON data
Mixed-type list elementslist[any]List of any type
Multi-level nestingdict[str, list[any]]Arbitrary depth

3.3 Type Conversion

CatBase provides the following type conversion functions:

Function Description
int(x) Convert x to an integer
float(x) Convert x to a float
str(x) Convert x to a string

Type Conversion Example


def main(args:list[str]) {
    # String to integer
    n:int = int("42")
    print("int('42') = ", n, "\n")
    
    # Integer to string
    s:str = str(123)
    print("str(123) = '", s, "'\n")
    
    # Integer to float
    f:float = float(10)
    print("float(10) = ", f, "\n")
    
    # String to float
    f2:float = float("3.14")
    print("float('3.14') = ", f2, "\n")
}

Output:


int('42') = 42
str(123) = '123'
float(10) = 10.0
float('3.14') = 3.14


3.4 Struct / Class

CatBase provides two equivalent keywords for defining user-defined composite types: struct and class. Both are completely equivalent in the language and differ only in style preference.

3.4.1 Keyword Selection

KeywordStyleTarget Audience
structC / Zig / Rust style, emphasizes "value-type aggregate"Systems programming background (Rust/Go/C programmers)
classPython / Java / JS style, emphasizes "object blueprint"Scripting language background (Python/Java/JS programmers)

3.4.2 Basic Usage (struct and class are equivalent)


# Declared with struct
struct Point {
    x: int
    y: int
}

# Declared with class (completely equivalent to above)
class Point {
    x: int
    y: int
}

def main(args: list[str]) {
    a: Point = Point { x: 1, y: 2 }
    b: Point = Point { x: 3, y: 4 }
    print(a.x, a.y)
}

3.4.3 Method Calls on Struct Fields & const struct Handling

CatBase allows you to call methods directly on struct fields, regardless of whether the field type is set / list / dict or another struct:


struct vst {
    key: set[str]
    value: set[str]
}

def main() {
    v: vst = vst { key: set(), value: set() }
    v.key.add("a")           # ← call set.add() on field `key`
    v.key.add("b")
    v.value.add("c")
    print(v)                 # vst{ key: {'a', 'b'}, value: {'c'} }
}

Under the hood (good to know, but the compiler handles it automatically):

  1. Prescan stage: when the compiler sees a method call (e.g. add / append / put) on a struct field such as v.key, it automatically promotes the struct variable v to var. Reason: in Zig, a const field is read-only, so methods taking a *T receiver are rejected.
  2. codegen stage: the receiver (e.g. v.key) is automatically wrapped with @constCast(&v.key), so v.key.add() compiles. Chained calls such as v.nested.field.method() are also supported.
  3. For typical scenarios, you don't need to think about any of this — just write v.key.add(...).

⚠️ Do NOT write @constCast(&v.key) yourself: the compiler will add it. Doing it manually is more error-prone.

Full example (covering nested set / list / dict / chained calls):


struct Data {
    ids: set[int]
    names: list[str]
    config: dict[str, any]
}

def main() {
    d: Data = Data {
        ids: set([1, 2, 3]),
        names: ["alice", "bob"],
        config: {"version": 1, "debug": false}
    }
    # call methods on struct fields
    d.ids.add(4)
    d.names.append("carol")
    d.config["max"] = 100
    # nested set() constructors automatically infer from field types
    d.ids.contains(1)            # true
    d.names.len()                # 3
    print(d)
}

Typical use case — struct fields of set type (social networks, tag systems, block lists, etc.):


struct Profile {
    friends: set[str]    # friends (auto-dedup)
    blocked: set[str]    # block list (auto-dedup)
}

def main() {
    alice: Profile = Profile {
        friends: set(["bob", "carol", "bob", "dave"]),    # duplicate "bob" auto-dedup
        blocked: set(["eve", "mallory"])
    }
    print(alice.friends)         # {'carol', 'dave', 'bob'}
    print(alice.blocked)         # {'eve', 'mallory'}

    # Call methods on fields directly: the compiler automatically
    # promotes alice to var and adds @constCast to alice.friends.
    alice.friends.add("frank")
    alice.friends.add("carol")   # duplicate, set already has it
    print(alice.friends)         # {'carol', 'dave', 'bob', 'frank'}

    # query APIs
    print(alice.friends.contains("bob"))    # true
    print(alice.friends.contains("ghost"))  # false
}

Multi-level nested struct + set (modeling hierarchical data):


struct Inner {
    tags: set[str]
}
struct Outer {
    name: str
    inner: Inner
}

def main() {
    box: Outer = Outer {
        name: "container",
        inner: Inner { tags: set(["alpha", "beta"]) }
    }
    print(box.name)              # container
    print(box.inner.tags)        # {'alpha', 'beta'}

    # chained access: box.inner.tags infers set[str] from Inner.tags
    box.inner.tags.add("gamma")
    print(box.inner.tags)        # {'alpha', 'gamma', 'beta'}
}

⚠️ Note: avoid using c or c.foo as variable names. CatBase treats c.xxx as a C library function call (e.g. c.malloc); using it as a struct field will fail to compile. Prefer semantically named variables such as box / data / cfg.

3.4.3.3 Edge Test Cases

Complete edge-case tests live in examples/test_struct_nested_set_edge.cat, covering six common scenarios:

ScenarioWhat it verifiesExample
0-arg set() initializationLHS context propagates through fieldsbox: Empty = Empty { items: set(), tags: set() }
Full set API on struct fieldsadd / len / contains / unionSet / intersection / differencealice.friends.contains("bob"), alice.friends.unionSet(other.friends)
3-level nesting (struct > list[struct] > set)chained access + method calls on list-element struct fieldsouter.middle.items[0].tags.add("a-new")
Mixed nesting (struct field is dict[str, set[int]])composition of multiple container typescache.entries: {"session-1": set([1,2,3])}
Large set + dedup50 elements added twice each still results in 50while i < 50 { big.numbers.add(i); big.numbers.add(i) }
Empty set operations∅ ∪ X = X, ∅ ∩ X = ∅set().unionSet(big_set).len() == big_set.len()

3-level nesting full code (verifies that CatBase compile-time type inference propagates through the list/struct chain):


struct Inner {
    tags: set[str]
}
struct Middle {
    name: str
    items: list[Inner]    # list elements may be struct, sidesteps the recursive struct restriction
}
struct Outer {
    title: str
    middle: Middle
}

def main() {
    box: Outer = Outer {
        title: "demo",
        middle: Middle {
            name: "section-A",
            items: [
                Inner { tags: set(["a", "b", "a"]) },     # duplicate "a" dedup
                Inner { tags: set(["c", "d"]) }
            ]
        }
    }
    print(box.middle.items.len())                  # 2
    print(box.middle.items[0].tags)                # {'a', 'b'}
    box.middle.items[0].tags.add("a-new")
    print(box.middle.items[0].tags)                # {'a', 'b', 'a-new'}
}

Note: CatBase forbids indirect recursive struct fields (e.g. Level1 { middle: Level2 { inner: Level3 } }), but allows list[Struct] as an intermediate layer to hold multiple instances.

Complete set method list (compatible with Python set):

MethodSignatureDescriptionPython equivalent
add(x)set[T] → boolAdd an elementset.add()
remove(x)set[T] → boolRemove an elementset.remove()
contains(x)set[T] → boolMembership checkx in s
len()set[T] → intElement countlen(s)
unionSet(other)set[T] → set[T]Union (named unionSet because union is a Zig reserved word)s | other
intersection(other)set[T] → set[T]Intersections & other
difference(other)set[T] → set[T]Differences - other

3.4.3.1 Struct Literals Must Include the Struct Name

A struct literal must include the struct name: StructName { field: value, ... }.

Anti-pattern (forgetting the struct name):


struct Point {
    x: int
    y: int
}

def main() {
    # ❌ Error: missing struct name "Point"
    v: Point = { x: 1, y: 2 }
}

The compiler emits a dedicated hint:


[Type Error] Variable assignment type mismatch
  Variable 'v' declared as 'Point' but assigned value of type 'dict[str,int]'
  hint: did you forget the struct name? Use `Point { ... }` instead of `{ ... }`
  example: `Point { x: <value>, y: <value> }`

Correct form:


def main() {
    # ✅ Correct: include the struct name explicitly
    v: Point = Point { x: 1, y: 2 }
    print(v)              # Point{ x: 1, y: 2 }
}

CatBase's { ... } syntax is by design a dict literal. Both Point { x: 1, y: 2 } and { x: 1, y: 2 } use {...}, but the Point prefix marks the literal as a user-defined type, telling the compiler whether {...} is a struct literal or a dict literal.

3.4.4 Core Concept: class is Syntax Sugar for ADT

class is syntax sugar for ADT (Algebraic Data Type), equivalent to struct, and does NOT include inheritance. Inheritance is a separate, independent language feature.

This explanation has solid theoretical support in programming language theory — C#, Rust, Swift, and Kotlin are all precedents:

LanguageSimilar ConceptInheritance?
C#struct (value type)❌ No
Ruststruct + impl block❌ (uses trait)
Swiftstruct (value type)❌ No
Kotlindata class❌ (final by default)
Gotype T struct { ... }❌ No
Zigconst T = struct { ... }❌ No

None of these 6 mainstream languages require their "class/struct" to support inheritance. The belief that "class must support inheritance" is a common misconception — in fact, when class was first introduced in a programming language (Simula 67, 1967), it did appear with inheritance, but the core meaning of "class" is "user-defined composite type (ADT)", and inheritance was a later optional addition.

3.4.4 Why CatBase Does Not Support Inheritance

CatBase chooses to give up inheritance in exchange for three core advantages:

  • Clear C ABI (FFI is simple and reliable)
    • struct/class memory layout is fully determined at compile time (no vtable, no RTTI)
    • CatBase-side struct and Zig-side struct have byte-for-byte layout correspondence
    • Zero overhead when passing structs across FFI boundaries
  • Explicit Field Layout (known at compile time)
    • All fields are determined at compile time
    • Memory size and field offsets are derivable from source
    • No runtime type information needed
  • Zero Overhead (no vtable, no RTTI)
    • Method calls use static dispatch (bound at compile time)
    • No dynamic dispatch overhead
    • No runtime type checking overhead

Why value types + inheritance is semantically unclear: if class B extends A, when a B-typed object is assigned to an A variable:

  • Trim B's fields? → Polymorphism breaks
  • Keep B's fields? → C ABI breaks (layout changes)
  • Adjust layout dynamically? → Both performance and ABI collapse

CatBase chose "clear C ABI" over "inheritance".

3.4.5 History of Inheritance

History of inheritance as a programming language feature:

YearLanguageInheritance Introduction
1967Simula 67First time "class" and inheritance were introduced together into a programming language
1972SmalltalkComplete OOP system (class + inheritance + dynamic dispatch)
1983C++Multiple inheritance + virtual function table (vtable)
1995JavaSingle inheritance + multiple interface implementation
2000C#Single inheritance + interface
2011Kotlinopen keyword controls inheritability (final by default)
2014SwiftSingle inheritance + protocol

Comparison with other languages' inheritance:

LanguageInheritance ModelvtableRTTIMultipleDefault Inheritable
C++Single/Multiple
JavaSingle❌ (uses interface)❌ (final by default)
C#Single❌ (uses interface)✅ (unless sealed)
PythonMultiple✅ (MRO)
RustNo class inheritanceN/A❌ (uses trait)N/A
ZigNo inheritanceN/A
GoNo class inheritance❌ (uses interface)N/A
CatBaseNo inheritanceN/A

3.4.6 Composition as an Alternative to Inheritance

CatBase does not support inheritance, but fully supports composition. The standard way to "reuse code" is composition:


class Animal {
    name: str
    age: int

    def describe() -> str {
        return self.name + " is " + str(self.age) + " years old"
    }
}

// Using composition for "Dog contains Animal" (instead of "Dog inherits Animal")
class Dog {
    animal: Animal      // Composition: Dog internally contains an Animal
    breed: str

    def describe() -> str {
        return self.animal.describe() + " (" + self.breed + ")"
    }
}

def make_dog(name: str, age: int, breed: str) -> Dog {
    return Dog {
        animal: Animal { name: name, age: age },
        breed: breed
    }
}

def main(args: list[str]) {
    dog: Dog = make_dog("Rex", 5, "Labrador")
    print(dog.describe())                          // Rex is 5 years old (Labrador)
    print("Dog's name:", dog.animal.name)          // Access composed object's field
}

Advantages of composition over inheritance:

  • Layout is controllable: each struct is independent, no vtable
  • More readable: dog.animal.name is more explicit than inherited dog.name (though more verbose)
  • Avoids the diamond inheritance problem: Python/Java/C++'s diamond problem simply cannot exist
  • Easier to test: you can mock Animal alone, without depending on an inheritance hierarchy

3.4.7 Practical Project Recommendations

Project TypeRecommended KeywordReason
New project (for Python/JS users)classLowers the entry barrier
New project (for systems programmers)structConsistent with Zig/Rust style
Cross-FFI projectstructConsistent with C ABI naming
Mixed teamPick one and stick with itAvoid inconsistent style within the team

Core principles:

  1. Use only one keyword within the same project (don't mix, to avoid confusing readers)
  2. Use struct for types that cross FFI (aligns with C ABI)
  3. Use class for purely internal CatBase types (more familiar)

3.4.8 FFI is Completely Unaffected

class in CatBase is pure syntax sugar and does not change the ABI:

DimensionstructclassSame?
Number of fieldsDetermined by definitionDetermined by definition✅ Identical
Memory sizeKnown at compile timeKnown at compile time✅ Identical
C ABI layoutstructstruct✅ Identical
Zig wrapper reception✅ Identical
Cross .so boundary✅ Identical

The Zig wrapper does not need to know whether the CatBase side uses class or struct — it only cares about the field layout.


3.5 None Type

Chapter Overview: None is CatBase's "no value" marker, similar to Python's None, C's NULL, or Java's null. It is a singleton value representing "no value" or "value unknown". This section explains how to use None, its type conversion rules, and how it interacts with the any type and dict.get().

3.5.1 Core Properties of None

  • None is a value, not a type: Like True and False, None is a concrete value that can be assigned to any variable whose type supports None.
  • Can be assigned to any type: int, str, float, bool, byte, bytes, any, list[T], and dict[K, V] can all be initialized with None to mean "this variable currently has no value".
  • Unified null check: Use is None / is not None to test whether a variable is empty.
  • Most useful with the any type: The any type can store None and is the workhorse type for JSON / configuration parsing scenarios.

3.5.2 Basic Usage of None

Example: Initializing variables with None


def main(args:list[str]) {
    # Any type can be initialized with None
    xint:int = None
    ystr:str = None
    xfloat: float = None
    xbool: bool = None
    xbyte: byte = None
    xbytes: bytes = None
    xany: any = None
    xlist: list[int] = None
    xdict: dict[str, str] = None

    # Use is None to check for emptiness
    if xint is None {print("xint is None")}
    if ystr is None {print("ystr is None")}
    if xfloat is None {print("xfloat is None")}
    if xbool is None {print("xbool is None")}
    if xbyte is None {print("xbyte is None")}
    if xbytes is None {print("xbytes is None")}
    if xany is None {print("xany is None")}
    if xlist is None {print("xlist is None")}
    if xdict is None {print("xdict is None")}

    # None values can be printed directly
    print(xint)        # None
    print(ystr)        # None
    print(xfloat)      # None
    print(xbool)       # None
    print(xbyte)       # None
    print(xbytes)      # None
    print(xany)        # None
    print(xlist)       # None
    print(xdict)       # None
}

Output:


xint is None
ystr is None
xfloat is None
xbool is None
xbyte is None
xbytes is None
xany is None
xlist is None
xdict is None
None
None
None
None
None
None
None
None
None

Example: Initialize with None, then assign real values


def main(args:list[str]) {
    xint:int = None
    ystr:str = None
    xfloat: float = None
    xbool: bool = None
    xbyte: byte = None
    xbytes: bytes = None
    xany: any = None
    xlist: list[int] = None
    xdict: dict[str, str] = None

    # Assign real values
    xint = 1
    ystr = "1"
    xfloat = 1
    xbool = True
    xbyte = b"A"
    xbytes = b"hello"
    xany = "a"
    xlist = [1, 2]
    xdict = {"b": "AAA"}

    # Use is not None to check whether a value has been assigned
    if xint is not None {print("xint is not None")}
    if ystr is not None {print("ystr is not None")}
    if xfloat is not None {print("xfloat is not None")}
    if xbool is not None {print("xbool is not None")}
    if xbyte is not None {print("xbyte is not None")}
    if xbytes is not None {print("xbytes is not None")}
    if xany is not None {print("xany is not None")}
    if xlist is not None {print("xlist is not None")}
    if xdict is not None {print("xdict is not None")}
}

Output:


xint is not None
ystr is not None
xfloat is not None
xbool is not None
xbyte is not None
xbytes is not None
xany is not None
xlist is not None
xdict is not None

Complete example: examples/test_none_check2.cat

3.5.3 Type Conversion Rules for None

None follows these rules when converted (consistent with Python):

ConversionResultDescription
int(None)0Empty value treated as integer 0
float(None)0.0Empty value treated as float 0.0
str(None)"None"String form (same as Python str(None))
bool(None)FalseEmpty value treated as false (same as Python)
bytes(None)empty bytesByte sequence of length 0
byte(None)0Single byte 0
bin(None)"0b0"bin(int(None))

3.5.4 None and dict.get()

Behavior of dict.get(key) when the key does not exist:

  • dict[str, any] (value type is any): returns None when not found
  • dict[str, T] (T is a concrete type such as int / str): returns the zero value of T (0 / "" / False), not None

def main(args:list[str]) {
    # dict[str, int] returns 0 when key not found
    data: dict[str, int] = {"b": 1}
    v: int = data.get("a")
    print(v)         # 0

    # dict[str, any] returns None when key not found
    any_data: dict[str, any] = {"name": "Alice", "age": 25}
    missing: any = any_data.get("nonexistent")
    if missing is None {
        print("key not found, value is None")
    }
}

See examples/test_dict_get_none.cat and examples/test_dict_get_none_2.cat.

3.5.5 None and the any Type

The any type can store None and is central to handling JSON / configuration data:

  • type(x) returns any:None for an any variable that holds None
  • Assigning None to an any variable means "no value"
  • When parsing JSON, null is automatically converted to None

def main(args: list[str]) {
    # Create an any variable, initialized to None
    x: any = None
    print("x (None) =", str(x), ", type =", type(x))   # x (None) = None , type = any:None

    # Fetch a missing key from dict[str, any]
    data: dict[str, any] = {"name": "Alice"}
    missing: any = data.get("missing")
    if missing is None {
        print("missing key returns None")              # hit
    }
}

3.5.6 None and Function Return Values

Functions without a declared return type implicitly return None:


def greet(name: str) {       # no return type
    print("Hello,", name)
    # implicitly returns None
}

def main(args: list[str]) {
    result: any = greet("CatBase")
    if result is None {
        print("greet() returned None")
    }
}

3.5.7 Notes on Using None

NoteDescription
Use is None / is not NoneDo not use == None / != None; use the is keyword for identity comparison (consistent with Python).
Do not confuse with 0 / ""None means "no value"; 0 / "" are valid values. Even though bool(None) == False, be explicit about the semantics.
Accessing a container set to None errorsBefore dereferencing a container that may be None, always check with is None to avoid null-pointer dereference errors.
Built-ins like int() / float()int(None) = 0 is the default behavior and does not throw (different from Python). If you need to distinguish, check for None before converting.

3.6 Notes on Using any Type Variables

any is CatBase's "universal container" type — flexible but with a cost. This section explains the usage boundaries of any variables, so you can avoid the most common mistakes before you make them.

3.6.1 Core Restriction: You Cannot Call .method() on any

This is the most common and easiest pitfall of the any type.

Local variables of type any are compiled as const in CatBase, so you cannot call methods (such as .get(), .append(), etc.) on them. The compiler will report an error and abort compilation.

Incorrect Example:


def main(args:list[str]) {
    raw: list[any] = [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]
    i: int = 0
    while i < len(raw) {
        nd: any = raw[i]                # ← declared as any
        name: any = nd.get("name", "")  # ← compile error: cannot call method on any
        print(name, "\n")
        i = i + 1
    }
}

Compiler Output (excerpt):


CatBase compilation errors:
  --> test_graph.cat:71
  |
     |   69 |             i: int = 0
     |   70 |             while i < len(raw) {
   > |   71 |                 nd: any = raw[i]
     |   72 |                 if int(str(nd.get("id", 0))) == target_id {
     |   73 |                     print("Node{id=", nd.get("id", 0), ", label=", nd.get("label", "")}\n")
  |
  [any-misuse-error] cannot call method on variable 'nd' of type 'any'.
    Hint: Declare 'nd' with a more specific type. For a dict-like value, use 'dict[str, any]'. For a list, use 'list[T]'. For a struct, use the concrete struct name.

The error is reported on the variable's declaration line (line 71: nd: any = raw[i]), and includes a hint about the recommended type.

3.6.2 Recommended Pattern: Use a More Specific Type

Whenever you know the real type of a value, you should avoid using any:

ScenarioUse instead
Dictionary parsed from JSONdict[str, any] (concrete container type, value can still be any)
Dict elements inside a listdict[str, any] (same as above)
Scalar values inside a listint / str / float / bool
Custom structured datastruct or class name

Correct Example:


def main(args:list[str]) {
    raw: list[dict[str, any]] = [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]
    i: int = 0
    while i < len(raw) {
        nd: dict[str, any] = raw[i]    # ← use concrete type dict[str, any]
        name: any = nd.get("name", "")  # ← .get() is now allowed
        print(name, "\n")
        i = i + 1
    }
}

3.6.3 Soft Warning: any Used as dict / list / struct

When an any variable does not call a method but its usage is clearly a dict / list / struct pattern (such as nd["key"] index access, .append() list method, struct field access, etc.), the compiler emits a soft warning ([any-usage]) suggesting a more specific type.

Soft warnings do not abort compilation, but you should fix them to improve type safety:


CatBase compilation warnings (non-fatal):
  --> test_graph.cat:191
  |
     |  189 |                 nd: any = nodes[i]
     |  190 |                 if int(str(nd.get("id", 0))) == target_id {
   > |  191 |                     nd["props"] = props
     |  192 |                     found = 1
  |
  [any-usage] variable 'nd' is declared as 'any' but accessed with dict pattern ("props")
    Hint: declare it as 'dict[K, V]' (e.g. 'dict[str, any]') so static type checks can verify key access

How to fix: change nd: any to nd: dict[str, any] and the warning (and any latent error) will go away.

3.6.4 Summary of Compiler Behavior

TriggerCompiler behavior
Direct method call any_var.method()Hard error (aborts compilation)
Index access any_var["key"] (dict pattern)Soft warning (does not abort)
List method any_var.append(...)Soft warning (does not abort)
Struct field access any_var.fieldSoft warning (does not abort)
Error message mentions "Zig"Never — CatBase intentionally hides low-level details
Does the low-level compiler still run when errors exist?No — CatBase exits immediately

3.6.5 Practical Rules of Thumb

  1. Prefer dict[str, any] over any when working with JSON / config / parsed data.
  2. Make list element types concrete: list[dict[str, any]] is more readable and produces fewer errors than list[any].
  3. If you really need a temporary any variable, only do assignment, str() / int() conversion, or is None checks on it — do not call methods.
  4. Fix [any-usage] warnings immediately: they almost always indicate a more specific type is appropriate.
  5. [any-misuse-error] always means failure: this code will never compile.

3.6.6 Side-by-side Comparison


# ===== Bad: calling method on any =====
def bad_example(data: list[any]) {
    item: any = data[0]
    return item.get("name", "")   # compile error
}

# ===== Warning: indexing into any =====
def warning_example(data: list[any]) {
    item: any = data[0]
    return item["name"]           # soft warning
}

# ===== Good: use a concrete type =====
def good_example(data: list[dict[str, any]]) {
    item: dict[str, any] = data[0]
    return item.get("name", "")   # OK
}

# ===== Flexible but controlled: any + conversion =====
def flex_example(data: list[any]) {
    item: any = data[0]
    return str(item)              # OK: built-in function, not a method
}

3.6.7 Correct Posture for is Type Guards

Key Point: After an is guard succeeds, the compiler automatically narrows any to a concrete type inside the then block; no additional int() / str() conversion is needed:


def main(args: list[str]) {
    x: any = 42

    # √ Recommended: use directly after guard
    if x is int {
        n: int = x          # Direct assignment to int, no int(x) needed
        print("n + 10 =", n + 10, "\n")
    }

    # × Redundant: converting again after the guard
    if x is int {
        n: int = int(x)     # Redundant conversion; compiler allows but it's unnecessary
    }
}

is not form:


def main(args: list[str]) {
    # Safe to use after checking "is not None"
    x: any = 42
    if x is not None {
        if x is int {
            print(x, "\n")
        }
    }
}

Chained multi-level guards:


def main(args: list[str]) {
    # Standard pattern for handling nested dict[str, any] structures
    data: dict[str, any] = {"user": {"name": "Alice", "age": 30}}

    user: any = data.get("user")
    if user is dict {
        name: any = user.get("name")
        if name is str {
            print("name:", str(name), "\n")
        }
    }
}

3.6.8 Choosing Between Explicit Conversion and Guards

The table below summarizes the applicable scenarios for both styles:

StyleApplicable ScenarioAdvantageDisadvantage
if x is T { use(x) }Type of x is uncertain; needs branchingType-safe, highly readableVerbose
str(x) / int(x)x is already a known typeOne-linerReturns default value (0 / empty string) for unexpected types

Practical Experience:

  1. JSON / config parsing → always use is guard first (value types are truly uncertain)
  2. Intermediate variables inside functions → use int() / str() directly (avoid unnecessary if nesting)
  3. Return value of dict.get() → use is guard before use
  4. Iterating mixed-type list → must use is guard inside the loop body

3.6.9 Best Practices for Container Nesting

JSON data access pattern:


def main(args: list[str]) {
    data: dict[str, any] = {
        "user": {"name": "Alice", "age": 30, "tags": ["admin", "user"]}
    }

    # Standard access pattern: get → guard → recurse
    if data.get("user") is dict {
        user: any = data.get("user")
        if user.get("tags") is list {
            tags: any = user.get("tags")
            if len(tags) > 0 {
                first: any = tags[0]
                if first is str {
                    print("first tag:", str(first), "\n")
                }
            }
        }
    }
}

Iterating list[any]:


def main(args: list[str]) {
    # Iterate a mixed-type list
    items: list[any] = [1, "a", True, None]
    for item in items {
        if item is int {
            print("int:", item, "\n")
        } else if item is str {
            print("str:", str(item), "\n")
        }
    }
}

Safe access for multi-level nested dict:


def get_nested_value(d: dict[str, any], path: list[str]) -> any {
    # Given a path list, drill down layer by layer
    current: any = d
    for key in path {
        if current is dict {
            current = current.get(key)
        } else {
            return None
        }
    }
    return current
}

def main(args: list[str]) {
    data: dict[str, any] = {"a": {"b": {"c": 42}}}
    result: any = get_nested_value(data, ["a", "b", "c"])
    if result is int {
        print("deep value:", result, "\n")   # 42
    }
}

3.6.10 Guidelines for using any in Function Signatures

Functions returning any — documentation must describe the possible concrete types:


# Return value can be str / int / None (caller must guard)
def lookup(key: str) -> any {
    if key == "name" {
        return "Alice"
    } else if key == "age" {
        return 30
    }
    return None
}

Functions with an any parameter — suitable for general-purpose utilities:


# Accept any type, print its type and value
def inspect(x: any) {
    print("type =", str(type(x)), ", value =", str(x), "\n")
}

Avoid unnecessary any parameters:

If you can enumerate all possible types, prefer multiple concrete-type overloads or a union type. Use any only when the value is truly "unknown".

3.6.11 Common Anti-patterns

Anti-pattern 1: Using any instead of dict[str, any]:


# × Anti-pattern
def bad(data: any) {
    return data.get("name")   # Error: cannot call method on any
}

# √ Correct
def good(data: dict[str, any]) {
    return data.get("name")   # OK
}

Anti-pattern 2: Converting back to any after a guard:


# × Anti-pattern
def bad(x: any) -> any {
    if x is int {
        return x          # Already narrowed to int; no need to convert back to any
    }
    return None
}

# √ Correct
def good(x: any) -> int {
    if x is int {
        return x
    }
    return 0
}

Anti-pattern 3: Calling methods without a guard:


# × Compile error
def bad(x: any) {
    return len(x)         # Error: any has no len() method
}

# √ Correct
def good(x: any) {
    if x is str {
        return len(x)     # len() works after guard
    }
    return 0
}

Anti-pattern 4: Expecting int() to parse any(str) without a guard:


# × Wrong expectation
def bad(x: any) {
    return int(x)         # If x is actually "123", returns 0 (does not parse the string)
}

# √ Correct
def good(x: any) {
    if x is str {
        return int(x)     # Only after the guard does int() parse the string
    }
    return 0
}

4. Operators

Chapter Overview: After mastering data types, this chapter introduces operators in CatBase. Operators are indispensable tools in programming, used to perform various calculations and operations on data. Through this chapter, you will be able to flexibly use arithmetic, comparison, logical, and other operators to build complex expressions.

4.1 Arithmetic Operators

Operator Description Example
+ Addition 5 + 3 → 8
- Subtraction 5 - 3 → 2
* Multiplication 5 * 3 → 15
/ Division 5 / 2 → 2
% Modulo 5 % 2 → 1

Arithmetic Operation Example


def main(args:list[str]) {
    a:int = 10
    b:int = 3
    
    print("a + b = ", a + b, "\n")
    print("a - b = ", a - b, "\n")
    print("a * b = ", a * b, "\n")
    print("a / b = ", a / b, "\n")
    print("a % b = ", a % b, "\n")
}

Output:


a + b = 13
a - b = 7
a \* b = 30
a / b = 3
a % b = 1

4.2 Compound Assignment Operators

CatBase supports compound assignment operators, which combine arithmetic operations with assignment into a single statement, making the code more concise.

Operator Description Equivalent Form Example
+= Add then assign a = a + b a += 3
-= Subtract then assign a = a - b a -= 3
*= Multiply then assign a = a * b a *= 3
/= Divide then assign a = a / b a /= 3

Compound Assignment Operation Example


def main(args:list[str]) {
    a:int = 10
    
    a += 5    # Equivalent to a = a + 5
    print("a += 5: ", a, "\n")
    
    a -= 3    # Equivalent to a = a - 3
    print("a -= 3: ", a, "\n")
    
    a *= 2    # Equivalent to a = a * 2
    print("a *= 2: ", a, "\n")
    
    a /= 4    # Equivalent to a = a / 4
    print("a /= 4: ", a, "\n")
}

Output:


a += 5: 15
a -= 3: 12
a *= 2: 24
a /= 4: 6

4.3 Relational Operators

Operator Description Example
== Equal to 5 == 5 → True
!= Not equal to 5 != 3 → True
< Less than 3 < 5 → True
> Greater than 5 > 3 → True
<= Less than or equal to 3 <= 5 → True
>= Greater than or equal to 5 >= 5 → True

Relational Operation Example


def main(args:list[str]) {
    a:int = 5
    b:int = 3
    
    print("a == b: ", a == b, "\n")
    print("a != b: ", a != b, "\n")
    print("a < b: ", a < b, "\n")
    print("a > b: ", a > b, "\n")
    print("a <= b: ", a <= b, "\n")
    print("a >= b: ", a >= b, "\n")
}

Output:


a == b: false
a != b: true
a < b: false
a > b: true
a <= b: false
a >= b: true

4.4 Logical Operators

Operator Description Example
and Logical AND True and False → False
or Logical OR True or False → True
not Logical NOT not True → False

Logical Operation Example


def main(args:list[str]) {
    a:bool = True
    b:bool = False
    
    print("a and b: ", a and b, "\n")
    print("a or b: ", a or b, "\n")
    print("not a: ", not a, "\n")
    print("not b: ", not b, "\n")
}

Output:


a and b: false
a or b: true
not a: false
not b: true

4.5 Membership Operators

Operator Description Example
in Check whether a key exists in a dictionary "name" in {"name": "Alice"} → True

in Operator Example

The in operator is used to check whether a specified key exists in a dictionary:


def main(args:list[str]) {
    data:dict[str,any] = {"name": "Alice", "age": 30}
    
    print("name in data: ", "name" in data, "\n")
    print("city in data: ", "city" in data, "\n")
    
    # Use in conditional statements
    if "age" in data {
        print("Age exists!\n")
    }
    
    # Combined with and
    if "name" in data and data["name"] {
        print("Name is not empty!\n")
    }
}

Output:


name in data: true
city in data: false
Age exists!
Name is not empty!

4.6 Operator Precedence

CatBase operator precedence (from high to low):

  1. () - Parentheses
  2. not - Logical NOT
  3. * / % - Multiplication, division, modulo
  4. + - - Addition, subtraction
  5. < > <= >= - Comparison
  6. == != - Equal to / not equal to
  7. in - Membership operator
  8. and - Logical AND
  9. or - Logical OR

Precedence Example


def main(args:list[str]) {
    # Multiplication and division take precedence over addition and subtraction
    print("2 + 3 * 4 = ", 2 + 3 * 4, "\n")
    
    # Parentheses can change precedence
    print("(2 + 3) * 4 = ", (2 + 3) * 4, "\n")
    
    # Logical operations
    print("not 1 > 2: ", not 1 > 2, "\n")
}

Output:


2 + 3 \* 4 = 14
(2 + 3) \* 4 = 20
not 1 > 2: true


5. Control Flow

Chapter Overview: After learning operators, this chapter introduces control flow statements. Control flow determines the execution order of a program, including conditional statements and loop statements. Mastering control flow allows you to write programs with branching and looping logic, enabling more complex functionality.

5.1 Conditional Statements

if Statement


def main(args:list[str]) {
    age:int = 18
    
    if age >= 18 {
        print("Adult\n")
    } else {
        print("Minor\n")
    }
}

Output:


Adult

if Variable (Simplified Empty Check)

CatBase supports using the simplified if variable syntax to check whether a string is empty:


def main(args:list[str]) {
    test_str:str = "hello"
    
    if test_str {
        print("String is not empty\n")
    }
    
    empty_str:str = ""
    if empty_str {
        print("String is not empty\n")
    } else {
        print("String is empty\n")
    }
}

Output:


String is not empty
String is empty

Syntax Description:

  • if variable - If the variable is not empty (string length > 0), the condition is true
  • Equivalent to the check if variable != ""

if-else if-else Statement


def main(args:list[str]) {
    score:int = 85
    
    if score >= 90 {
        print("Excellent\n")
    } else if score >= 80 {
        print("Good\n")
    } else if score >= 60 {
        print("Pass\n")
    } else {
        print("Fail\n")
    }
}

Output:


Good

Nested if Statements


def main(args:list[str]) {
    x:int = 10
    y:int = 20
    
    if x > 0 {
        if y > 0 {
            print("Both x and y are positive\n")
        } else {
            print("x is positive, y is negative\n")
        }
    } else {
        print("x is negative\n")
    }
}

Output:


Both x and y are positive

5.2 Loop Statements

for Loop


def main(args:list[str]) {
    # Iterate over a list
    nums:list[int] = [1, 2, 3, 4, 5]
    
    for i in nums {
        print("Value: ", i, "\n")
    }
}

Output:


Value: 1
Value: 2
Value: 3
Value: 4
Value: 5

for Loop with Index


def main(args:list[str]) {
    fruits:list[str] = ["apple", "banana", "orange"]
    
    for i, v in fruits {
        print(i, ": ", v, "\n")
    }
}

Output:


0: apple
1: banana
2: orange

Range for Loop


def main(args:list[str]) {
    # Iterate over a range
    for i in range(5) {
        print("i = ", i, "\n")
    }
}

Output:


i = 0
i = 1
i = 2
i = 3
i = 4

for Loop (Iterator)

CatBase supports using the for ... in syntax to iterate over objects that implement iterator methods, such as HTTP streaming responses.

To use an iterator loop, the following conditions must be met:

  1. The object implements the iter_lines() method
  2. The method returns an iterator that yields one line of content per iteration
  3. The loop ends automatically when the iterator returns None

def main(args:list[str]) {
    # Send an HTTP request (streaming mode)
    url:str = "http://localhost:19090/v1/chat/completions"
    data:dict[str,any] = {
        "model": "qwen",
        "messages": [{"role": "user", "content": "Hello"}],
        "stream": true
    }
    
    # Use stream=True to enable streaming mode
    response:Response = http_post(url, json=data, stream=True)
    
    # Check response status
    response.raise_for_status()
    
    # Iterate to read the streaming response
    for line in response.iter_lines() {
        print(line)
    }
}

Output:


data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"Hello"},"logprobs":null,"finish_reason":null}]}
data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"!"},"logprobs":null,"finish_reason":null}]}
data: [DONE]

Description:

  • The stream=True parameter enables streaming mode, returning a Response object instead of a string
  • response.raise_for_status() checks the HTTP status code; any value outside the 200-299 range triggers an exception
  • response.iter_lines() returns an iterator that yields one line of response content per iteration
  • The loop ends automatically when the iterator returns None
Parsing Streaming Responses

When handling streaming API responses, you typically need to parse the JSON data of each line:


def main(args:list[str]) {
    url:str = "http://localhost:19090/v1/chat/completions"
    data:dict[str,any] = {
        "model": "qwen",
        "messages": [{"role": "user", "content": "Hello"}],
        "stream": true
    }
    
    response:Response = http_post(url, json=data, stream=True)
    response.raise_for_status()
    
    # Parse the streaming response
    for line in response.iter_lines() {
        line_str:str = line
        if line_str.startswith("data: ") {
            # Extract the JSON part
            data_json:str = line_str[6:]
            
            # Parse JSON
            data_dict:dict = json_loads(data_json)
            
            # Check the choices field
            if "choices" in data_dict and data_dict["choices"] {
                # Get the content from delta
                delta:dict = data_dict["choices"][0].get("delta", {})
                content:any = delta.get("content", "")
                
                # Convert to string and print
                content_str:str = str(content)
                print(content_str, end="", flush=True)
            }
        }
    }
}

Output:


Hello! How can I help you?

Parsing Steps Description:

  1. Use startswith("data: ") to check whether it is a data line
  2. Use slice [6:] to extract the JSON string
  3. Use json_loads() to parse JSON into a dictionary
  4. Use the in operator to check whether a key exists
  5. Use the .get() method to safely retrieve nested values
  6. Use str() to convert Variant to a string
  7. Use the end="" parameter to print without a newline

while Loop


def main(args:list[str]) {
    i:int = 0
    
    while i < 5 {
        print("i = ", i, "\n")
        i = i + 1
    }
}

Output:


i = 0
i = 1
i = 2
i = 3
i = 4

break


def main(args:list[str]) {
    # break example
    print("Break example:\n")
    for i in range(10) {
        if i == 5 {
            break
        }
        print(i, " ")
    }
    print("\n")
}

Output:


Break example:
0 1 2 3 4

Note: CatBase currently does not support the continue keyword. To skip certain iterations, use an if conditional statement instead.

5.3 Exception Handling

CatBase uses try...except statements to handle exceptions. The exception handling mechanism allows you to catch and handle runtime errors in your code, making programs more robust.

Basic Syntax

CatBase supports the following except syntax:

  1. No-variable syntax (recommended): except { ... } - no need to specify an exception variable
  2. Variable syntax: except errname - capture the exception into a variable
  3. Typed syntax: except Exception as errname - capture the exception into a variable

Note: The legacy catch keyword is also compatible, with the same usage as except.

except Without a Variable (Recommended Syntax)


def main(args:list[str]) {
    try {
        print("hello\n")
    }
    except {
        print("something went wrong\n")
    }
}

Output:


hello

except with an Exception Variable


def main(args:list[str]) {
    try {
        print("hello\n")
    }
    except e {
        print("Caught error: ")
        print(e)
    }
}

Output:


hello

except with Type and Variable


def main(args:list[str]) {
    try {
        print("hello\n")
    }
    except Exception as e {
        print("Caught exception: ")
        print(e)
    }
}

Output:


hello

Nested try-except


def main(args:list[str]) {
    try {
        try {
            print("inner try\n")
        }
        except Exception as e {
            print("inner except: ")
            print(e)
        }
    }
    except Exception as e {
        print("outer except: ")
        print(e)
    }
}

Output:


inner try

Exception Handling Rules

  1. Unified exception type: All exceptions are of type Exception; there is no distinction between error and Exception
  2. Exception variable scope: The exception variable (such as e or err) is only visible inside the except block
  3. Catch anywhere: try-except can be used anywhere in a function
  4. Keyword choice: except is recommended, but catch is also compatible

Printing Exception Information

In the except block, you can use print(errname) to directly print the details of the exception:


def main(args:list[str]) {
    try {
        x:int = 10
        y:int = 0
        if y == 0 {
            print("Error: Division by zero\n")
        }
    }
    except Exception as e {
        print("Caught: ")
        print(e)
    }
}

Output:


Error: Division by zero
Caught: error.RuntimeError

Compatibility with the Legacy catch Syntax

CatBase is also compatible with the legacy catch keyword:


def main(args:list[str]) {
    try {
        print("hello\n")
    }
    catch {
        print("caught\n")
    }
}

Output:


hello

finally Block

The finally block is used to define code that will execute regardless of whether an exception occurs. It is typically used for resource cleanup, such as closing files or releasing locks.


def main(args:list[str]) {
    try {
        print("try block\n")
    }
    except {
        print("catch block\n")
    }
    finally {
        print("finally block - always executes\n")
    }
}

Output:


try block
finally block - always executes

Even when an exception occurs, finally still executes:


def main(args:list[str]) {
    try {
        print("try block - about to error\n")
        # Simulate an error here
    }
    except {
        print("catch block\n")
    }
    finally {
        print("finally block - cleanup here\n")
    }
}

Output:


try block - about to error
catch block
finally block - cleanup here

Typical scenarios for using finally for resource cleanup:


def main(args:list[str]) {
    stream: RecordStream = recordStream(rate=16000, channels=1)

    try {
        data:bytes = record(5, "16000", "1", "", "1024")
        save_wav(data, "/tmp/recording.wav", "16000")
    }
    except {
        print("Error during recording\n")
    }
    finally {
        # Ensure the recording stream is closed and resources are released
        stream.close()
        print("Resources cleaned up\n")
    }
}

6. Functions

Chapter Overview: Once you have mastered sequential execution and conditional logic, this chapter introduces functions. Functions are the basic unit for organizing code, improving reusability and readability. CatBase functions are designed to be concise and easy to use, with support for return values, making your programs more modular.

6.1 Function Definition and Invocation

Basic Function Definition


def greet(name:str) {
    print("Hello, ", name, "!\n")
}

def main(args:list[str]) {
    greet("CatBase")
}

Output:


Hello, CatBase!

Function with Return Value


def add(a:int, b:int) -> int {
    return a + b
}

def main(args:list[str]) {
    result:int = add(5, 3)
    print("5 + 3 = ", result, "\n")
}

Output:


5 + 3 = 8

Multiple Return Values

CatBase supports two ways to declare functions that return lists:

Method 1: Explicitly specify the element type


def divide() -> list[int] {
    quotient:int = 1
    remainder:int = 2
    r:list[int]=[quotient, remainder]
    return r
}

def main(args:list[str]) {
    result:list[int] = divide()
    print(result[0],result[1])
}

Method 2: Implicitly infer the element type


def divide() -> list {
    quotient:int = 1
    remainder:int = 2
    r:list=[quotient, remainder]
    return r
}

def main(args:list[str]) {
    result:list = divide()
    print(result[0],result[1])
}

Output:


1 2

Design Principle: The CatBase compiler automatically performs type inference. When a function's return type is declared as list (without specifying the element type), the compiler infers and fills in the concrete element type (such as list[int]) based on the actual list literal or variable type returned in the return statement. Similarly, when a variable is declared as list but assigned a list of a concrete type, the variable type is automatically updated to the concrete type.

Usage Notes:

>

- Explicit declaration (such as list[int]) provides better code readability and stricter type checking

- Implicit inference (such as list) is more flexible, but it is recommended to ensure all return paths in the function return lists of the same type

Complete Example: Both Methods Used Together


# Method 1: Explicitly specify the element type
def get_list1() -> list[int] {
    return [1, 2, 3, 4, 5]
}

# Method 2: Implicitly infer the element type
def get_list2() -> list {
    return [1, 2, 3, 4, 5]
}

def main(args:list[str]) {
    result1:list[int] = get_list1()
    result2:list = get_list2()
    print("Method 1: ", result1, "\n")
    print("Method 2: ", result2, "\n")
}

Output:


Method 1:  [1, 2, 3, 4, 5] 
Method 2:  [1, 2, 3, 4, 5] 

Dictionary as Return Value Example


# Method 1: Explicitly specify key-value types
def get_dict1() -> dict[str, int] {
    d:dict[str, int] = {"a": 1, "b": 2, "c": 3}
    return d
}

# Method 2: Implicitly infer key-value types
def get_dict2() -> dict {
    d:dict[str, int] = {"x": 10, "y": 20, "z": 30}
    return d
}

def main(args:list[str]) {
    result1:dict[str, int] = get_dict1()
    result2:dict = get_dict2()
    print("Method 1: ", result1, "\n")
    print("Method 2: ", result2, "\n")
    print("result1[\"a\"] = ", result1["a"], "\n")
    print("result2[\"y\"] = ", result2["y"], "\n")
}

Output:


Method 1:  {'a': 1, 'b': 2, 'c': 3} 
Method 2:  {'x': 10, 'y': 20, 'z': 30} 
result1["a"] =  1 
result2["y"] =  20 

Note: Dictionaries, like lists, also support implicit type inference. When a function's return type is declared as dict (without specifying key-value types), the compiler automatically infers the concrete key-value types based on the dictionary variable type in the return statement.

6.2 Recursive Functions


def factorial(n:int) -> int {
    if n <= 1 {
        return 1
    }
    return n * factorial(n - 1)
}

def main(args:list[str]) {
    print("5! = ", factorial(5), "\n")
    print("10! = ", factorial(10), "\n")
}

Output:


5! = 120
10! = 3628800

6.3 Function Parameters

Default Parameters


def greet(name:str, greeting:str) {
    print(greeting, ", ", name, "!\n")
}

def main(args:list[str]) {
    greet("Tom", "Hello")
    greet("Jerry", "Hi")
}

Output:


Hello, Tom!
Hi, Jerry!

Function References as Parameters

CatBase supports passing function references as parameters to other functions, making callback patterns possible.


def on_audio_frame(frame:bytes) {
    print("Received audio frame")
    print(len(frame))
}

def process_audio(callback:function) {
    # The callback function will be invoked
    callback(bytes("test data"))
}

def main(args:list[str]) {
    # Pass the on_audio_frame function as a parameter
    process_audio(on_audio_frame)
}

Output:


Received audio frame
9

Description:

  • When a defined function name is passed as a parameter, it is treated as the function type
  • Function references can be used for callback-style APIs, such as audio processing, network communication, and other scenarios
  • Function references cannot be printed directly (Zig does not support formatting function pointers)

6.4 Variable Scope


def main(args:list[str]) {
    x:int = 10
    
    print("Outer x = ", x, "\n")
    
    {
        x:int = 20
        print("Inner x = ", x, "\n")
    }
    
    print("Outer x after block = ", x, "\n")
}

Output:


Outer x = 10
Inner x = 20
Outer x after block = 10


7. Built-in Functions

Chapter Overview: After learning about custom functions, this chapter introduces CatBase's built-in functions. Built-in functions are common utilities provided by the language that can be used directly without definition, greatly improving development efficiency. CatBase provides a rich set of built-in functions covering type conversion, mathematical operations, string processing, and more.

CatBase provides a rich set of built-in functions.

7.1 Print Function

print

print(...) - Print values to standard output


def main(args:list[str]) {
    print("Hello\n")
    print("Number: ", 42, "\n")
    print("Bool: ", True, "\n")
}

Output:


Hello
Number: 42
Bool: true

print Function Parameter Description

The print() function supports the following parameters:

  • Positional arguments: The values to print; multiple values can be provided
  • end: The ending character, defaults to a newline \n; set to an empty string "" to avoid a newline
  • flush: Whether to flush output, defaults to False; set to True to flush output immediately

def main(args:list[str]) {
    # Print without a newline
    print("Hello ", end="", flush=True)
    print("World!", end="", flush=True)
    
    # Use the default newline
    print("\nDone!")
}

Output:


Hello World!
Done!

7.2 Math Functions

abs

abs(x:int) : int - Returns the absolute value


def main(args:list[str]) {
    print("abs(-5) = ", abs(-5), "\n")
    print("abs(5) = ", abs(5), "\n")
    print("abs(0) = ", abs(0), "\n")
}

Output:


abs(-5) = 5
abs(5) = 5
abs(0) = 0

max

max(a:int, b:int) : int - Returns the larger value


def main(args:list[str]) {
    print("max(5, 3) = ", max(5, 3), "\n")
    print("max(-10, 10) = ", max(-10, 10), "\n")
}

Output:


max(5, 3) = 5
max(-10, 10) = 10

min

min(a:int, b:int) : int - Returns the smaller value


def main(args:list[str]) {
    print("min(5, 3) = ", min(5, 3), "\n")
    print("min(-10, 10) = ", min(-10, 10), "\n")
}

Output:


min(5, 3) = 3
min(-10, 10) = -10

sum

sum(list) : int - Returns the sum of all elements in a list


def main(args:list[str]) {
    nums:list[int] = [1, 2, 3, 4, 5]
    print("sum([1,2,3,4,5]) = ", sum(nums), "\n")
    print("sum([]) = ", sum([]), "\n")
}

Output:


sum([1,2,3,4,5]) = 15
sum([]) = 0

pow

pow(base:int, exp:int) : int - Returns the result of exponentiation


def main(args:list[str]) {
    print("pow(2, 3) = ", pow(2, 3), "\n")
    print("pow(5, 0) = ", pow(5, 0), "\n")
    print("pow(10, 2) = ", pow(10, 2), "\n")
}

Output:


pow(2, 3) = 8
pow(5, 0) = 1
pow(10, 2) = 100

round

round(x:int|float) : int|float - Rounds to the nearest integer, returns the same type as the input round(x:int|float, n:int) : float - Rounds to the specified number of decimal places, returns float


def main(args:list[str]) {
    print("round(3.7) = ", round(3.7), "\n")
    print("round(3.2) = ", round(3.2), "\n")
    print("round(3.5) = ", round(3.5), "\n")
    print("round(3.14159, 2) = ", round(3.14159, 2), "\n")
}

Output:


round(3.7) = 4
round(3.2) = 3
round(3.5) = 4
round(3.14159, 2) = 3.14

7.3 Type Conversion Functions

int

int(x) : int - Convert to integer


def main(args:list[str]) {
    print("int('42') = ", int("42"), "\n")
    print("int(3.7) = ", int(3.7), "\n")
    print("int(True) = ", int(True), "\n")
}

Output:


int('42') = 42
int(3.7) = 3
int(True) = 1

float

float(x) : float - Convert to float


def main(args:list[str]) {
    print("float(10) = ", float(10), "\n")
    print("float('3.14') = ", float("3.14"), "\n")
}

Output:


float(10) = 10.0
float('3.14') = 3.14

str

str(x) : str - Convert to string


def main(args:list[str]) {
    print("str(123) = '", str(123), "'\n")
    print("str(3.14) = '", str(3.14), "'\n")
    print("str(True) = '", str(True), "'\n")
}

Output:


str(123) = '123'
str(3.14) = '3.14'
str(True) = 'true'

str Function and JSON Value Conversion

The str() function can also convert values (of type dict) parsed from JSON into strings. After parsing a JSON string with json_loads(), the values in the resulting dictionary may be of the Variant type; use str() to convert them to strings:


def main(args:list[str]) {
    # Parse a JSON string
    json_str:str = "{\"name\": \"Alice\", \"age\": 30, \"score\": 95.5}"
    data:dict = json_loads(json_str)
    
    # Extract values and convert to strings
    name:str = str(data["name"])
    age:str = str(data["age"])
    score:str = str(data["score"])
    
    print("Name: ", name, "\n")
    print("Age: ", age, "\n")
    print("Score: ", score, "\n")
}

Output:


Name: Alice
Age: 30
Score: 95.5

The str() function supports converting the following Variant types:

  • .str - String
  • .int - Integer
  • .float - Float
  • .bool - Boolean (converted to "true" or "false")
  • .null - JSON null value (converted to "null", corresponding to JSON standard null)
  • .list - List (converted to a JSON array string)
  • .dict - Dictionary (converted to a JSON object string)

Note: In CatBase code, the null value is represented by None (corresponding to JSON's null). For example:


def main(args:list[str]) {
    # Use None to represent a null value in CatBase
    value:any = None
    
    # Check whether it is null
    if value == None {
        print("Value is None\n")
    }
}

### 7.4 Base Conversion Functions

#### bin

`bin(x:int) : str` - Convert to binary

def main(args:list[str]) {
    print("bin(5) = '", bin(5), "'\n")
    print("bin(10) = '", bin(10), "'\n")
    print("bin(255) = '", bin(255), "'\n")
}

Output:


bin(5) = '0b101'
bin(10) = '0b1010'
bin(255) = '0b11111111'

oct

oct(x:int) : str - Convert to octal


def main(args:list[str]) {
    print("oct(8) = '", oct(8), "'\n")
    print("oct(10) = '", oct(10), "'\n")
    print("oct(64) = '", oct(64), "'\n")
}

Output:


oct(8) = '0o10'
oct(10) = '0o12'
oct(64) = '0o100'

hex

hex(x:int) : str - Convert to hexadecimal


def main(args:list[str]) {
    print("hex(255) = '", hex(255), "'\n")
    print("hex(16) = '", hex(16), "'\n")
    print("hex(4096) = '", hex(4096), "'\n")
}

Output:


hex(255) = '0xff'
hex(16) = '0x10'
hex(4096) = '0x1000'

7.5 Character Functions

chr

chr(x:int|byte) : str - Convert an integer (code point) to a single-character string. Accepts int or byte.


def main(args:list[str]) {
    print("chr(65) = '", chr(65), "'\n")
    print("chr(97) = '", chr(97), "'\n")
    print("chr(48) = '", chr(48), "'\n")
    # Also accepts byte
    b:byte = byte(90)
    print("chr(byte(90)) = '", chr(b), "'\n")
}

Output:


chr(65) = 'A'
chr(97) = 'a'
chr(48) = '0'
chr(byte(90)) = 'Z'

Note: chr() treats the integer (0-255) as a Unicode/ASCII code point and returns a single-character string. This is different from str(); see the comparison table below.

ord

ord(x:str) : int - Convert a character (first byte) to an integer (code point)


def main(args:list[str]) {
    print("ord('A') = ", ord("A"), "\n")
    print("ord('a') = ", ord("a"), "\n")
    print("ord('0') = ", ord("0"), "\n")
}

Output:


ord('A') = 65
ord('a') = 97
ord('0') = 48

Key Difference Between chr/ord and str/int

CatBase provides two different sets of conversion functions. Understanding their distinction is critical for proper use:

Inputchr(x) / ord(x)str(x) / int(x)
65chr(65) = "A" (1 char)str(65) = "65" (2 chars)
10chr(10) = "\n" (newline)str(10) = "10" (chars "1" "0")
"A"ord("A") = 65 (single char)int("A") compile error

chr() and ord(): bidirectional char ↔ integer conversion

  • chr(x): integer → single-character string (treats the integer as a code point)
  • ord(s): single-character string → integer (returns the first byte's code point)
  • Use cases: character encoding, ASCII code processing, character arithmetic

str() and int(): value to/from string

  • str(x): any value → numeric string (e.g. str(65) = "65")
  • int(s): string → integer (e.g. int("42") = 42)
  • Use cases: displaying numbers, parsing strings, JSON serialization

byte type conversion:

  • byte(65): construct a byte value (u8, 0-255)
  • chr(byte(65)) = "A": byte takes the chr path
  • Currently str(byte) is not supported (different semantics from str(int)). For a "numeric string" form, convert to int first: str(int(b))

def main(args:list[str]) {
    # char ↔ integer
    c1:str = chr(65)        # "A"
    n1:int = ord("Z")       # 90
    print(c1, " has code ", n1, "\n")

    # string ↔ integer
    s1:str = str(65)        # "65"
    n2:int = int("42")      # 42
    print(s1, " + ", n2, " = ", int(s1) + n2, "\n")

    # byte type
    b:byte = byte(255)
    print("byte(255) as char: ", chr(b), "\n")
}

7.6 String Functions

len

len(x) : int - Returns the length


def main(args:list[str]) {
    print("len('hello') = ", len("hello"), "\n")
    print("len([1,2,3]) = ", len([1, 2, 3]), "\n")
    print("len({'a':1}) = ", len({"a": 1}), "\n")
}

Output:


len('hello') = 5
len([1,2,3]) = 3
len({'a':1}) = 1

range

range(n:int) : list - Generate a range


def main(args:list[str]) {
    print("range(5) = ", range(5), "\n")
    print("range(2, 5) = ", range(2, 5), "\n")
}

Output:


range(5) = [0, 1, 2, 3, 4]
range(2, 5) = [2, 3, 4]

String Methods

String objects provide a rich set of methods for manipulation. Below are all available string methods:

Case Conversion

def main(args:list[str]) {
    s:str = "Hello World"
    print("Original string: ", s, "\n")
    print("upper(): ", s.upper(), "\n")
    print("lower(): ", s.lower(), "\n")
    print("capitalize(): ", s.capitalize(), "\n")
    print("title(): ", s.title(), "\n")
}

Output:


Original string: Hello World
upper(): HELLO WORLD
lower(): hello world
capitalize(): Hello world
title(): Hello World

Whitespace Handling

def main(args:list[str]) {
    s:str = "   Hello World   "
    print("Original string: '", s, "'\n")
    print("strip(): '", s.strip(), "'\n")
    print("lstrip(): '", s.lstrip(), "'\n")
    print("rstrip(): '", s.rstrip(), "'\n")
}

Output:


Original string: '   Hello World   '
strip(): 'Hello World'
lstrip(): 'Hello World   '
rstrip(): '   Hello World'

Trimming / Padding (trim / center / ljust / rjust / zfill)

CatBase provides string methods 100% compatible with Python. trim() with no arguments is equivalent to strip() (auto-removes leading/trailing whitespace \t\n\r). center() / ljust() / rjust() default to space as fillchar when omitted.


def main() {
    s:str = "  Hello, World!  "
    print("s.trim()        = '", s.trim(), "'")
    print("s.trim(' !')    = '", s.trim(" !"), "'")

    c:str = "hi"
    print("c.center(6, '-')  = '", c.center(6, "-"), "'")
    print("c.ljust(5, '.')   = '", c.ljust(5, "."), "'")
    print("c.rjust(5, '.')   = '", c.rjust(5, "."), "'")
    print("'42'.zfill(5)     = '", "42".zfill(5), "'")
    print("'-42'.zfill(5)    = '", "-42".zfill(5), "'")
}

Output:


s.trim()        = 'Hello, World!'
s.trim(' !')    = 'Hello, World'
c.center(6, '-')  = '--hi--'
c.ljust(5, '.')   = 'hi...'
c.rjust(5, '.')   = '...hi'
'42'.zfill(5)     = '00042'
'-42'.zfill(5)    = '-0042'
Advanced String Operations (partition / removeprefix / swapcase / expandtabs)

CatBase also provides advanced string methods 100% compatible with Python:


def main() {
    rp:str = "TestHook"
    print("'TestHook'.removeprefix('Test') = '", rp.removeprefix("Test"), "'")
    print("'TestHook'.removesuffix('Hook') = '", rp.removesuffix("Hook"), "'")

    p:str = "hello world hello"
    parts:list[str] = p.partition("world")
    print("partition('world') = ", parts)
    parts3:list[str] = p.rpartition("hello")
    print("rpartition('hello') = ", parts3)

    print("'HELLO'.casefold()  = '", "HELLO".casefold(), "'")
    print("'Hello'.swapcase()  = '", "Hello".swapcase(), "'")
    print("'a\\tb'.expandtabs(4) = '", "a\tb".expandtabs(4), "'")
}

Output:


'TestHook'.removeprefix('Test') = 'Hook'
'TestHook'.removesuffix('Hook') = 'Test'
partition('world') = ["hello ","world"," hello"]
rpartition('hello') = ["hello world ","hello",""]
'HELLO'.casefold()  = 'hello'
'Hello'.swapcase()  = 'hELLO'
'a\tb'.expandtabs(4) = 'a   b'

Note: CatBase's str is a byte array (UTF-8 encoded). casefold() only handles ASCII characters. For full Unicode folding, use an external library.

Search and Count

def main(args:list[str]) {
    s:str = "Hello World Hello"
    print("Original string: ", s, "\n")
    print("find('World'): ", s.find("World"), "\n")
    print("rfind('Hello'): ", s.rfind("Hello"), "\n")
    print("count('l'): ", s.count("l"), "\n")
    print("startswith('Hello'): ", s.startswith("Hello"), "\n")
    print("endswith('World'): ", s.endswith("World"), "\n")
}

Output:


Original string: Hello World Hello
find('World'): 6
rfind('Hello'): 12
count('l'): 3
startswith('Hello'): true
endswith('World'): false

String Slicing

CatBase supports string slicing operations to extract substrings:


def main(args:list[str]) {
    s:str = "Hello World"
    
    # From index 6 to the end
    result1:str = s[6:]
    print("s[6:]: ", result1, "\n")
    
    # From index 0 to index 5 (exclusive)
    result2:str = s[0:5]
    print("s[0:5]: ", result2, "\n")
    
    # From index 6 to index 11
    result3:str = s[6:11]
    print("s[6:11]: ", result3, "\n")
}

Output:


s[6:]:  World
s[0:5]: Hello
s[6:11]: World

Syntax Description:

  • str[start:] - From the start index to the end of the string
  • str[start:end] - From the start index to the end index (end exclusive)
  • Indexes start at 0
Split and Join

def main(args:list[str]) {
    s:str = "apple,banana,cherry"
    print("Original string: ", s, "\n")
    result:list = s.split(",")
    print("split(','): ", result, "\n")
    
    s2:str = "Hello\nWorld\n!"
    print("Original string: ", s2, "\n")
    result2:list = s2.splitlines()
    print("splitlines(): ", result2, "\n")
    
    items:list[str] = ["Hello", "World"]
    joined:str = ",".join(items)
    print("join(): ", joined, "\n")
}

Output:


Original string: apple,banana,cherry
split(','): [apple, banana, cherry]
Original string: Hello
World
!
splitlines(): [Hello, World, !]
join(): Hello,World

Replace

def main(args:list[str]) {
    s:str = "Hello World"
    print("Original string: ", s, "\n")
    print("replace('World', 'CatBase'): ", s.replace("World", "CatBase"), "\n")
}

Output:


Original string: Hello World
replace('World', 'CatBase'): Hello CatBase

Character Checks

def main(args:list[str]) {
    print("'hello'.islower(): ", "hello".islower(), "\n")
    print("'HELLO'.isupper(): ", "HELLO".isupper(), "\n")
    print("'Hello'.istitle(): ", "Hello".istitle(), "\n")
    print("'123'.isdigit(): ", "123".isdigit(), "\n")
    print("'abc'.isalpha(): ", "abc".isalpha(), "\n")
    print("'abc123'.isalnum(): ", "abc123".isalnum(), "\n")
    print("'   '.isspace(): ", "   ".isspace(), "\n")
    print("'123'.isnumeric(): ", "123".isnumeric(), "\n")
}

Output:


'hello'.islower(): true
'HELLO'.isupper(): true
'Hello'.istitle(): true
'123'.isdigit(): true
'abc'.isalpha(): true
'abc123'.isalnum(): true
'   '.isspace(): true
'123'.isnumeric(): true

7.7 System Functions

exec

exec(cmd:str) : str - Execute a system command


def main(args:list[str]) {
    result:str = exec("echo hello")
    print("exec result: ", result, "\n")
}

Output:


exec result: hello

sleep

sleep(seconds:int) - Pause execution


def main(args:list[str]) {
    print("Before sleep\n")
    sleep(1)
    print("After sleep\n")
}

Output:


Before sleep
After sleep

time

time() - Get the current timestamp (milliseconds), returns a float


def main(args:list[str]) {
    t:float = time()
    print("Current timestamp: ")
    print(t)
}

Output:


Current timestamp: 1717412345.0

perf_counter

perf_counter() - Get a high-resolution counter (nanoseconds), returns a float, used for precise measurement of code execution time


def main(args:list[str]) {
    start:float = perf_counter()
    # Simulate a time-consuming operation
    s:int = 0
    for i in range(1000000) {
        s = s + i
    }
    end:float = perf_counter()
    print("Elapsed: ")
    print((end - start) / 1000000000.0, " seconds")
}

Output:


Elapsed: 0.012345 seconds

strftime

strftime(format:str) : str - Format the time as a string, returns a format like "2024-06-04 12:00:00"


def main(args:list[str]) {
    s:str = strftime("%Y-%m-%d %H:%M:%S")
    print("Current time: ")
    print(s)
}

Output:


Current time: 2024-06-04 12:00:00

localtime

localtime() - Get the local time, returns a time struct containing year, month, day, hour, minute, second, weekday fields


def main(args:list[str]) {
    t:TimeStruct = localtime()
    print("Year: ")
    print(t.year)
    print("Month: ")
    print(t.month)
    print("Day: ")
    print(t.day)
    print("Hour: ")
    print(t.hour)
    print("Minute: ")
    print(t.minute)
    print("Second: ")
    print(t.second)
}

Output:


Year: 2024
Month: 6
Day: 4
Hour: 12
Minute: 0
Second: 30

gmtime

gmtime() - Get UTC time, returns a time struct containing year, month, day, hour, minute, second, weekday fields


def main(args:list[str]) {
    t:TimeStruct = gmtime()
    print("UTC Year: ")
    print(t.year)
    print("UTC Month: ")
    print(t.month)
    print("UTC Day: ")
    print(t.day)
    print("UTC Hour: ")
    print(t.hour)
}

Output:


UTC Year: 2024
UTC Month: 6
UTC Day: 4
UTC Hour: 4

timestamp_to_struct

timestamp_to_struct(timestamp:float) : TimeStruct - Convert a timestamp to a time struct


def main(args:list[str]) {
    ts:float = time()
    t:TimeStruct = timestamp_to_struct(ts)
    print("Year: ")
    print(t.year)
    print("Month: ")
    print(t.month)
    print("Day: ")
    print(t.day)
}

Output:


Year: 2024
Month: 6
Day: 4

strftime_timestamp

strftime_timestamp(timestamp:float, format:str) : str - Convert a timestamp to a string according to the specified format


def main(args:list[str]) {
    ts:float = time()
    s:str = strftime_timestamp(ts, "%Y-%m-%d %H:%M:%S")
    print("Formatted: ")
    print(s)
}

Output:


Formatted: 2024-06-04 12:00:00

mktime

mktime(t:TimeStruct) : float - Convert a time struct to a timestamp


def main(args:list[str]) {
    t:TimeStruct = localtime()
    ts:float = mktime(t)
    print("Timestamp: ")
    print(ts)
}

Output:


Timestamp: 1717412345.0

input

input(prompt:str) : str - Read user input from standard input


def main(args:list[str]) {
    name:str = input("Please enter your name: ")
    print("Hello, ", name, "!\n")
}

Output:


Please enter your name: CatBase
Hello, CatBase!

type

type(x) : str - Returns the type name of a variable


def main(args:list[str]) {
    a:int = 10
    b:str = "hello"
    c:list[int] = [1, 2, 3]
    d:dict[str, str] = {"key": "value"}
    
    print("type(10) = ", type(a), "\n")
    print("type('hello') = ", type(b), "\n")
    print("type([1,2,3]) = ", type(c), "\n")
    print("type(dict) = ", type(d), "\n")
}

Output:


type(10) = int
type('hello') = str
type([1,2,3]) = list
type(dict) = dict

For any type variables, type() returns an any:actual_type format, indicating the variable is of any type and showing the actual type of its currently stored value:


def main(args:list[str]) {
    x: any = None
    print("type(None) = ", type(x), "\n")

    x = 42
    print("type(42) = ", type(x), "\n")

    x = "hello"
    print("type('hello') = ", type(x), "\n")

    x = 3.14
    print("type(3.14) = ", type(x), "\n")

    x = True
    print("type(True) = ", type(x), "\n")

    x = [1, 2, 3]
    print("type(list) = ", type(x), "\n")

    x = {"key": "value"}
    print("type(dict) = ", type(x), "\n")
}

Output:


type(None) = any:None
type(42) = any:int
type('hello') = any:str
type(3.14) = any:float
type(True) = any:bool
type(list) = any:list
type(dict) = any:dict

assert

assert(condition:bool, message:str) - Assert; terminates the program and outputs an error message when the condition is false

assert is used to set checkpoints in code; when a condition is not met, the program terminates immediately. This is very useful during debugging and development, allowing problems to be caught early.


def divide(a:int, b:int) -> int {
    # Assert: the divisor cannot be 0
    assert(b != 0, "Division by zero")
    return a / b
}

def main(args:list[str]) {
    result:int = divide(10, 2)
    print("10 / 2 = ", result, "\n")
    
    # The following line would trigger an assertion failure
    # result = divide(10, 0)
}

Output:


10 / 2 = 5

Description:

  • assert accepts two parameters: a condition expression and an error message string
  • When the condition is False, the program terminates and outputs the error message
  • When the condition is True, the program continues normal execution
  • Recommended for use during debugging and development to verify the correctness of program logic

isinstance

isinstance(x, type_name:str) : bool - Check whether a variable is of the specified type


def main(args:list[str]) {
    a:int = 10
    b:str = "hello"
    c:list[int] = [1, 2, 3]
    
    print("isinstance(10, 'int') = ", isinstance(a, "int"), "\n")
    print("isinstance(10, 'str') = ", isinstance(a, "str"), "\n")
    print("isinstance('hello', 'str') = ", isinstance(b, "str"), "\n")
    print("isinstance([1,2,3], 'list') = ", isinstance(c, "list"), "\n")
}

Output:


isinstance(10, 'int') = true
isinstance(10, 'str') = false
isinstance('hello', 'str') = true
isinstance([1,2,3], 'list') = true

7.8 Pointer Type

CatBase provides the Pointer type for handling pointer parameters when calling external libraries (.so/.a). This is particularly useful when calling C libraries, especially when library functions need to modify the values of passed-in variables.

Pointer Type Overview

The Pointer type is a wrapper around Zig's ?*anyopaque, an optional opaque pointer type. It is primarily used for:

  • Calling external C functions that require pointer parameters
  • Modifying variable values through pointers
  • Handling pointer data returned by external libraries

Creating Pointers

pointer() - Create a null pointer


def main(args:list[str]) {
    # Create a null pointer
    empty: Pointer = pointer()
    
    # Check whether it is a null pointer
    if empty.is_null() {
        print("Pointer is null\n")
    }
}

pointer_of(var) - Create a pointer to a variable


def main(args:list[str]) {
    # Create an integer variable
    num: int = 42

    # Create a pointer to num
    ptr: Pointer = pointer_of(num)

    # Get the value through the pointer (multi-type: type is required)
    val: int = ptr.get(int)
    print("Value: ", val, "\n")

    # Set a new value through the pointer (multi-type: type is required)
    ptr.set(int, 100)
    print("After set: ", num, "\n")
}

Pointer Methods

Method Description
ptr.is_null() Check whether the pointer is null (returns bool)
ptr.get(int) Get the int value pointed to by the pointer (multi-type support)
ptr.get(float) Get the float value pointed to by the pointer
ptr.get(str) Get the str value pointed to by the pointer
ptr.get(bool) Get the bool value pointed to by the pointer
ptr.set(int, value) Set the int value pointed to by the pointer
ptr.set(float, value) Set the float value pointed to by the pointer
ptr.set(str, value) Set the str value pointed to by the pointer
ptr.set(bool, value) Set the bool value pointed to by the pointer
ptr.toPtrPtr() Get a pointer to the internal ptr field ([*c][*c]u8), used for output pointer parameters of C functions like sqlite3_open

Multi-Type Support (v0.0.7+)

The Pointer type supports accessing data of any primitive type through a comptime type parameter. The syntax is:


# Read data: ptr.get(type)
v: int = ptr.get(int)
v2: float = ptr.get(float)
s: str = ptr.get(str)

# Write data: ptr.set(type, value)
ptr.set(int, 100)
ptr.set(float, 3.14)
ptr.set(str, "hello")

Type Mapping Table:

CatBase Type Zig Type
inti64
floatf64
boolbool
strruntime.Str
byteu8
i8/i16/i32/i64i8/i16/i32/i64
u8/u16/u32/u64u8/u16/u32/u64
f32/f64f32/f64

Multi-Type Usage Example:


def main(args: list[str]) {
    # int type
    num: int = 42
    ptr_int: Pointer = pointer_of(num)
    print(ptr_int.get(int), "\n")
    ptr_int.set(int, 100)

    # float type
    f: float = 3.14
    ptr_float: Pointer = pointer_of(f)
    print(ptr_float.get(float), "\n")
    ptr_float.set(float, 2.71)

    # str type
    s: str = "hello"
    ptr_str: Pointer = pointer_of(s)
    print(ptr_str.get(str), "\n")
    ptr_str.set(str, "world")

    # Null pointer safety
    empty: Pointer = pointer()
    v_int: int = empty.get(int)
    v_float: float = empty.get(float)
}

Backward Compatibility: For compatibility with old code, the following shorthand forms are still supported (defaulting to int):


v: int = ptr.get()     # Equivalent to ptr.get(int)
ptr.set(100)              # Equivalent to ptr.set(int, 100)

Application Scenarios

Scenario 1: Calling C Functions That Modify External Variables

Some C functions need to return multiple values through pointers:


# Suppose there is a C function that returns the calculation result through pointers
# void calculate(int input, int *result, int *remainder)
# Both result and remainder are output parameters

import "./libcalc.so" as calc

# Declare the external function
from calc import calculate(input: int, result: int, remainder: int) -> None

def main(args:list[str]) {
    result: int = 0
    remainder: int = 0
    
    # Create pointers to result and remainder
    result_ptr: Pointer = pointer_of(result)
    remainder_ptr: Pointer = pointer_of(remainder)
    
    # Call the function (here you need to pass the address via Pointer)
    # Note: The actual invocation depends on the library's API design
}
Scenario 2: Handling Pointers Returned by External Libraries

import "./libdata.so" as data_lib

# Declare external functions
from data_lib import get_buffer_size() -> int
from data_lib import read_buffer(buf: Pointer, size: int) -> int

def main(args:list[str]) {
    # Get the buffer size
    size: int = data_lib.get_buffer_size()
    
    # Create a buffer
    buffer: bytes = bytes(" " * size)
    
    # Read data into the buffer
    bytes_read: int = data_lib.read_buffer(buffer, size)
    
    print("Read ", bytes_read, " bytes\n")
}
Scenario 3: Using the Pointer Type When Declaring External Functions

When an external function parameter is a pointer type, you can use Pointer directly in the from...import declaration:


import "./libserial.so" as serial_lib

# Declare external functions; Pointer is used for scenarios that require pointer parameters
from serial_lib import serial_write(handle: int, data: Pointer, len: int) -> int
from serial_lib import serial_read(handle: int, data: Pointer, len: int) -> int
from serial_lib import serial_close(handle: int) -> None

def main(args:list[str]) {
    # Open the serial port (assuming it returns a handle)
    handle: int = serial_lib.serial_open("/dev/ttyUSB0", 115200)
    
    # Prepare the data
    msg: str = "Hello, Serial!"
    msg_bytes: bytes = bytes(msg)
    
    # Write the data
    written: int = serial_lib.serial_write(handle, pointer_of(msg_bytes), len(msg_bytes))
    print("Written ", written, " bytes\n")
    
    # Close the serial port
    serial_lib.serial_close(handle)
}

Notes

  • pointer_of() automatically obtains the address of the passed-in variable so that its value can be modified through the pointer
  • When using ptr.set(value), the value type should be int (CatBase's int type maps to i64 at the underlying level)
  • Calling get() or set() on a null pointer will cause the program to panic
  • Use is_null() to safely check whether a pointer is null

7.9 Serial Communication Functions

CatBase provides complete serial communication support. You can use the serial() constructor to create a serial port object, then read, write, and close it through object methods.

Creating a Serial Connection

serial(port, baud_rate)

Create a serial port object and open the serial device.

Parameters:

  • port: Serial device path
    • Linux/macOS: such as /dev/ttyUSB0, /dev/ttyS0
    • Windows: such as COM1, COM2
  • baud_rate: Baud rate, supports 300, 1200, 2400, 4800, 9600, 19200, 38400, 57600, 115200, 230400

Returns: Serial serial port object


def main(args:list[str]) {
    # Open the serial port
    ser:Serial = serial("/dev/ttyUSB0", 115200)
    print("Serial port opened successfully\n")

    # Write data
    ser.write("Hello, Serial!\n")

    # Read data
    data:str = ser.read(1024)
    print("Received: ")
    print(data)
    print("\n")

    # Close the serial port
    ser.close()
    print("Serial port closed\n")
}

Serial Object Methods

Method Description
ser.write(data) Write data to the serial port (data is str or bytes)
ser.read(length) Read up to length bytes of data from the serial port
ser.available() Check the number of bytes available to read from the serial port
ser.close() Close the serial port connection

Serial Bytes Send/Receive Details

In serial communication, the ser.write() and ser.read() methods can accept and return string types, while the bytes type is essentially a special string that can be used directly for serial send/receive. Below are specific applications of bytes in serial communication:

Sending Binary Data

def main(args:list[str]) {
    ser:Serial = serial("/dev/ttyUSB0", 115200)

    # Send a Modbus RTU read register command
    # Frame format: device address(1) + function code(1) + start address(2) + register count(2) + CRC(2)
    modbus_read:bytes = b"\x01\x03\x00\x00\x00\x0A"
    ser.write(modbus_read)

    # Send a custom protocol data packet
    # Header + length + data + checksum
    header:bytes = b"\xAA\x55"
    length:bytes = b"\x00\x08"
    data:bytes = b"\x01\x02\x03\x04\x05\x06\x07\x08"
    checksum:bytes = b"\x2C"

    packet:bytes = header + length + data + checksum
    ser.write(packet)

    print("Binary data sent\n")
    ser.close()
}
Receiving and Parsing Binary Data

def main(args:list[str]) {
    ser:Serial = serial("/dev/ttyUSB0", 115200)

    # Send a read request
    ser.write(b"\x01\x03\x00\x00\x00\x0A")

    # Wait for data to arrive (poll up to 100 times)
    i:int = 0
    while i < 100 {
        avail:int = ser.available()
        if avail >= 8 {
            # Modbus response is at least 8 bytes: address + function code + length + data(4 bytes) + CRC(2 bytes)
            data:str = ser.read(avail)

            # Parse the response data
            # Assume the received data format is: \x01\x03\x04\x00\x00\x00\x01\xXX\xXX
            #                     address  function code  length   data high byte  data low byte  CRC
            print("Received data\n")
            print("Data length: ")
            print(len(data))
            print("\n")
            break
        }
        i = i + 1
    }

    ser.close()
}
Complete Binary Protocol Communication Example

def main(args:list[str]) {
    ser:Serial = serial("/dev/ttyUSB0", 115200)
    print("Serial port opened\n")

    # Define the protocol frame format
    # | Header(2) | Type(1) | Length(2) | Data(N) | Checksum(1) |
    # | 0xAA 0x55 | 0x01 | 0x0008 | N bytes | XOR |

    # Send a sensor query command
    frame_type:bytes = b"\x01"
    data_length:bytes = b"\x00\x08"
    sensor_cmd:bytes = b"\x01\x02\x03\x04\x05\x06\x07\x08"

    # Calculate the XOR checksum
    xor_sum:bytes = b"\x00"
    j:int = 0
    while j < len(sensor_cmd) {
        # Simple checksum calculation (actual protocols may be more complex)
        j = j + 1
    }

    # Assemble the complete frame
    header:bytes = b"\xAA\x55"
    frame:bytes = header + frame_type + data_length + sensor_cmd + xor_sum

    # Send the frame
    ser.write(frame)
    print("Sent ")
    print(len(frame))
    print(" bytes\n")

    # Receive the response
    i:int = 0
    while i < 200 {
        avail:int = ser.available()
        if avail > 0 {
            response:str = ser.read(avail)
            print("Received ")
            print(len(response))
            print(" bytes: ")
            print(response)
            print("\n")

            # Verify the header
            if len(response) >= 2 {
                # Check whether it starts with 0xAA 0x55
                print("Protocol verified\n")
            }
            break
        }
        i = i + 1
    }

    ser.close()
    print("Serial port closed\n")
}
Hex Escape Description

In a bytes literal, \xHH represents a hexadecimal byte value:

Literal Actual Byte Value Description
\x00 0 Null byte
\x01 - \x0F 1-15 Control characters
\x41 65 ASCII 'A'
\xFF 255 Maximum value

Example:


def main(args:list[str]) {
    # Hex string "41 42 43" (ABC)
    hex_str:bytes = b"\x41\x42\x43"
    print(hex_str)
    print("\n")
}

8. File Operations

Chapter Overview: After mastering functions and built-in functions, this chapter introduces file operations. File operations are a very common requirement in actual development, used for persistent data storage. CatBase provides a concise file operation API for convenient file reading and writing.

8.1 Opening and Closing Files

file

file(filename:str, mode:str) : File - Open file

Parameters:

  • filename - File name
  • mode - Open mode:
    • "r" - Read mode
    • "w" - Write mode (overwrite)
    • "a" - Append mode
    • "r+" - Read-write mode (can read and write, does not create a new file)

def main(args:list[str]) {
    f:File = file("test.txt", "w")
    f.write("Hello, World!")
    f.close()
    
    f2:File = file("test.txt", "r")
    content:str = f2.read()
    print("Content: ", content, "\n")
    close(f2)
}

⚠ File Mode and File Region Locking

To use the lock / tryLock / unlock methods from 8.6 File Region Locking, the file must be opened in "r+" mode. Other modes will fail write-lock attempts because the file descriptor lacks write permission (tryLock will always return false).

ModeCan lock?Description
"r"Read-only; fcntl(F_WRLCK) returns EBADF
"w"Write (truncate)
"a"Append
"r+"Read/write (recommended for locking scenarios)

# Correct: use "r+" mode
f:File = file("data.bin", "r+")
ok:bool = f.tryLock(0, 10)   # returns true or false normally

# Wrong: use "r" mode
f:File = file("data.bin", "r")
ok:bool = f.tryLock(0, 10)   # always returns false

8.2 Reading Files

read

file.read() : str - Read the entire file


def main(args:list[str]) {
    f:File = file("test.txt", "r")
    content:str = f.read()
    print("Content: ", content, "\n")
    f.close()
}

8.3 Writing Files

write

file.write(content:str) - Write to file (overwrite mode)


def main(args:list[str]) {
    f:File = file("output.txt", "w")
    f.write("Line 1\n")
    f.write("Line 2\n")
    f.write("Line 3\n")
    f.close()
    
    print("File written successfully\n")
}

append

file.append(content:str) - Append to file

Appends content to the end of the file without overwriting existing content.


def main(args:list[str]) {
    # First write
    f:File = file("log.txt", "w")
    f.write("Log entry 1\n")
    f.close()
    
    # Append write
    f2:File = file("log.txt", "a")
    f2.append("Log entry 2\n")
    f2.append("Log entry 3\n")
    close(f2)
    
    # Read to verify
    f3:File = file("log.txt", "r")
    content:str = f3.read()
    close(f3)
    
    print("File content:\n", content, "\n")
}

Run result:


File content:
Log entry 1
Log entry 2
Log entry 3

writeAt

file.writeAt(content:str, position:int) - Write at specified position

Writes content at the specified position (byte offset) in the file.


def main(args:list[str]) {
    # Create a file with fixed-length content
    f:File = file("data.txt", "w")
    f.write("AAAAAAAAAAAA")  # 12 characters
    f.close()
    
    # Write "BBB" at position 5 (replace characters)
    f2:File = file("data.txt", "r+")
    f2.writeAt("BBB", 5)
    f2.close()
    
    # Read to verify
    f3:File = file("data.txt", "r")
    content:str = f3.read()
    close(f3)
    
    print("File content: ", content, "\n")
}

Run result:


File content: AAAAABBBAAA

💡 writeAt + File Region Locking = Concurrency-Safe Writing

writeAt is not concurrency-safe by itself. If two processes/threads call writeAt on the same file region concurrently, they will overwrite each other and corrupt data.

Combine with 8.6 File Region Locking to write safely: lock → write → unlock.


def safe_write(f:File, offset:int, len:int, data:str) {
    # 1. Lock (only one process can hold the region at a time)
    f.tryLock(offset, len)

    # 2. Write under the lock
    f.writeAt(data, offset)

    # 3. Release the lock
    f.unlock(offset, len)
}

def main(args:list[str]) {
    f:File = file("data.bin", "r+")

    # Concurrently write to different regions of the file safely
    safe_write(f, 0, 10,  "AAAAAAAAAA")  # write to [0, 10)
    safe_write(f, 10, 10, "BBBBBBBBBB")  # write to [10, 20)
    safe_write(f, 20, 10, "CCCCCCCCCC")  # write to [20, 30)

    f.close()
}

See 8.6 File Region Locking and examples/test_file_lock.cat for a complete multi-threaded demonstration.

close

close(file:File) - Close file

Closes the opened file and releases related resources.


def main(args:list[str]) {
    f:File = file("test.txt", "w")
    f.write("Hello")
    f.close()  # Close file
}

Note: After completing file operations, you must call close() to close the file.

💡 close() automatically releases all locks on the file

If the file holds any file region locks (8.6 File Region Locking), calling close() automatically releases all locks on that file descriptor; no explicit unlock is required.


f:File = file("data.bin", "r+")
f.lock(0, 10)        # lock [0, 10)
f.lock(20, 10)       # lock [20, 30)
f.close()            # close file, automatically releasing both locks

So you can safely omit unlock before close(). However, explicit unlock is still good practice (see 8.6) as it makes the lock release timing more explicit.

8.4 Complete Example


def main(args:list[str]) {
    # Write file
    f:File = file("demo.txt", "w")
    f.write("Hello, CatBase!")
    f.close()
    
    # Read file
    f2:File = file("demo.txt", "r")
    content:str = f2.read()
    close(f2)
    
    print("Read from file: ", content, "\n")
}

Run result:


Read from file: Hello, CatBase!

8.5 File Operations and Bytes

File operations combined with the bytes type can be used to handle binary files, such as images, audio, and compressed files.

Writing Binary Data

def main(args:list[str]) {
    # Create binary data to write
    # Simulate writing a simple BMP image file header (54 bytes)
    # BMP file header
    bmp_header:bytes = b"BM"                    # File identifier
    file_size:bytes = b"\x00\x00\x00\x00"      # File size (placeholder)
    reserved:bytes = b"\x00\x00\x00\x00"        # Reserved field
    offset:bytes = b"\x36\x00\x00\x00"          # Pixel data offset (54)

    # DIB header (40 bytes)
    dib_size:bytes = b"\x28\x00\x00\x00"        # DIB header size
    width:bytes = b"\x10\x00\x00\x00"           # Width (16)
    height:bytes = b"\x10\x00\x00\x00"           # Height (16)
    planes:bytes = b"\x01\x00"                  # Number of color planes
    bits_per_pixel:bytes = b"\x18\x00"           # Bits per pixel (24)
    compression:bytes = b"\x00\x00\x00\x00"     # Compression method
    image_size:bytes = b"\x00\x00\x00\x00"      # Image size
    x_pixels_per_m:bytes = b"\x00\x00\x00\x00"  # Horizontal resolution
    y_pixels_per_m:bytes = b"\x00\x00\x00\x00"  # Vertical resolution
    colors_used:bytes = b"\x00\x00\x00\x00"     # Colors used
    colors_important:bytes = b"\x00\x00\x00\x00" # Important colors

    # Combine the complete BMP file header
    header:bytes = bmp_header + file_size + reserved + offset + dib_size + width + height + planes + bits_per_pixel + compression + image_size + x_pixels_per_m + y_pixels_per_m + colors_used + colors_important

    # Write to file
    f:File = file("test.bmp", "wb")
    f.write(header)
    f.close()

    print("BMP header written (54 bytes)\n")
}
Reading Binary Files

def main(args:list[str]) {
    # Read file in binary mode
    f:File = file("test.bmp", "rb")
    data:str = f.read()
    f.close()

    # Analyze BMP file header
    print("File size: ")
    print(len(data))
    print(" bytes\n")

    # Check file identifier (BM = 0x42 0x4D)
    if len(data) >= 2 {
        print("File identifier: ")
        print(data[0])
        print(data[1])
        print("\n")
    }

    # Read width info (offset 18, 4 bytes)
    if len(data) >= 22 {
        print("Width info in header\n")
    }
}
Writing and Reading Custom Binary Format

def main(args:list[str]) {
    # Create data packet
    # Format: | Magic(2) | Version(1) | Length(2) | Data(N) | CRC(4) |

    magic:bytes = b"\xCA\xTB"          # Magic number
    version:bytes = b"\x01"            # Version 1
    payload:bytes = b"Hello, Binary!"  # Data payload
    length:bytes = b"\x00\x0F"        # Length 15

    # Calculate simple CRC checksum
    crc:bytes = b"\x00\x00\x00\x00"

    # Assemble packet
    packet:bytes = magic + version + length + payload + crc

    # Write to binary file
    f:File = file("data.bin", "wb")
    f.write(packet)
    f.close()

    print("Packet written: ")
    print(len(packet))
    print(" bytes\n")

    # Read binary file
    f2:File = file("data.bin", "rb")
    received:str = f2.read()
    close(f2)

    # Verify packet
    if len(received) >= 5 {
        # Check magic number
        print("Received packet, length: ")
        print(len(received))
        print("\n")

        # Extract data payload
        if len(received) > 9 {
            payload_len:int = len(received) - 9
            print("Payload length: ")
            print(payload_len)
            print("\n")
        }
    }
}
Binary Protocol File Storage

def main(args:list[str]) {
    # Store multiple binary records to file
    # Record format: | ID(4) | Type(1) | Data Length(2) | Data(N) |

    f:File = file("records.bin", "wb")

    # Record 1
    id1:bytes = b"\x00\x00\x00\x01"
    type1:bytes = b"\x01"
    data1:bytes = b"\x41\x42\x43\x44"      # ABCD
    len1:bytes = b"\x00\x04"
    record1:bytes = id1 + type1 + len1 + data1

    # Record 2
    id2:bytes = b"\x00\x00\x00\x02"
    type2:bytes = b"\x02"
    data2:bytes = b"\x01\x02\x03\x04\x05"   # 5 bytes of data
    len2:bytes = b"\x00\x05"
    record2:bytes = id2 + type2 + len2 + data2

    # Write all records
    f.write(record1)
    f.write(record2)
    f.close()

    print("Records written to file\n")

    # Read and parse records
    f2:File = file("records.bin", "rb")
    content:str = f2.read()
    close(f2)

    # Parse record 1
    print("Record 1 ID: ")
    print(len(content))
    print("\n")
}
Appending Binary Data

def main(args:list[str]) {
    # Write binary data in append mode
    f:File = file("log.bin", "ab")

    # Write binary log entry
    # Format: | Timestamp(8) | Type(1) | Data(N) |
    timestamp:bytes = b"\x00\x00\x00\x00\x00\x00\x00\x01"
    entry_type:bytes = b"\x01"
    log_data:bytes = b"\xDE\xAD\xBE\xEF"

    entry:bytes = timestamp + entry_type + log_data
    f.write(entry)
    f.close()

    print("Binary log entry appended\n")

    # Read binary log
    f2:File = file("log.bin", "rb")
    log_content:str = f2.read()
    close(f2)

    print("Total log size: ")
    print(len(log_content))
    print(" bytes\n")
}

8.6 File Region Locking

File region locks (also called record locks) allow locking specific byte ranges of a file, enabling multiple processes (or different threads in the same process) to safely access different regions of the same file concurrently.

Core API

CatBase provides three lock methods on the File class:

Method Signature Returns Behavior
tryLock f.tryLock(offset: int, len: int) -> bool bool Non-blocking lock attempt: returns true on success, false if the region is already held
lock f.lock(offset: int, len: int) None Blocking lock: waits until the lock becomes available
unlock f.unlock(offset: int, len: int) None Releases a lock previously acquired by lock or tryLock

Parameter Description:

  • offset: the starting byte offset of the locked region (0-based)
  • len: the byte length of the locked region (0 means "from offset to end of file")

Platform Implementation:

  • POSIX (Linux/macOS): via fcntl(F_SETLK) / fcntl(F_SETLKW) system calls
  • Windows: via LockFileEx / UnlockFile Win32 APIs

Basic Example


def main(args:list[str]) {
    # Prepare a test file
    f:File = file("/tmp/data.bin", "w")
    f.write("ABCDEFGHIJKLMNOPQRSTUVWXYZ")  # 26 bytes
    f.close()

    # Open for read/write ("r+" is required; otherwise fcntl write lock will fail)
    f:File = file("/tmp/data.bin", "r+")

    # 1. Non-blocking lock attempt
    ok:bool = f.tryLock(0, 10)   # attempt to lock region [0, 10)
    if ok {
        print("Lock acquired\n")
        f.writeAt("1234567890", 0)
        f.unlock(0, 10)          # release with the same offset/len
    } else {
        print("Region is held by another process\n")
    }

    # 2. Different regions can be locked simultaneously
    f2:File = file("/tmp/data.bin", "r+")
    ok2:bool = f2.tryLock(20, 6)  # region [20, 26) does not overlap with [0, 10)
    if ok2 {
        f2.writeAt("XYZ", 20)
        f2.unlock(20, 6)
    }

    f.close()
    f2.close()
}

Blocking Lock

lock() is the blocking variant — it waits until the lock becomes available:


def writer(id:int, path:str) {
    f:File = file(path, "r+")
    f.lock(0, 16)
    print("Writer ", id, " got the lock\n")
    f.writeAt("I am writer " + str(id), 0)
    sleep(1)
    f.unlock(0, 16)
    f.close()
}

def main(args:list[str]) {
    f_init:File = file("/tmp/shared.bin", "w")
    f_init.write("0000000000000000")
    f_init.close()

    t1:Thread = thread writer(1, "/tmp/shared.bin")
    t2:Thread = thread writer(2, "/tmp/shared.bin")
    t1.join()
    t2.join()
}

Important Notes

  1. Advisory lock

    CatBase's file region locks are advisory. The OS does not enforce them; all collaborators must call lock / tryLock explicitly to prevent conflicts.

  2. POSIX locks are per-process

    The same process opening the same file with different fds will not block itself (POSIX spec). To achieve real cross-process mutual exclusion, use different processes or different threads.

  3. Release with the same offset/len

    
    f.lock(0, 10)
    f.unlock(0, 10)    # correct
    f.unlock(0, 20)    # wrong — the lock is not released
    
  4. Locks are released when the file is closed

    
    f.lock(0, 10)
    f.close()  # automatically releases the [0, 10) lock
    
  5. The file must be opened in "r+" mode

    lock / tryLock / unlock need a writable file descriptor. If you open with "r" (read-only), fcntl will return EBADF and tryLock will always return false.

    
    f:File = file("data.txt", "r+")  # correct
    f:File = file("data.txt", "r")   # wrong — locking will fail
    
  6. len = 0 means "from offset to end of file"

    This is a POSIX-standard special semantic:

    
    f.lock(100, 0)  # locks the region [100, end of file)
    

Comparison with Other Languages

Feature CatBase Python C
Lock type advisory (POSIX fcntl) advisory (fcntl) mandatory/advisory
Non-blocking tryLock → bool fcntl.flock(f, LOCK_EX \| LOCK_NB) fcntl(fd, F_SETLK, ...)
Blocking lock fcntl.flock(f, LOCK_EX) fcntl(fd, F_SETLKW, ...)
Release unlock fcntl.flock(f, LOCK_UN) fcntl(fd, F_SETLK, F_UNLCK)

Complete Example

See examples/test_file_lock.cat for a complete multi-threaded lock demonstration with 5 scenarios:

  1. Non-blocking tryLock
  2. Different regions can be locked simultaneously
  3. Safe write while holding the lock
  4. Re-locking after unlock
  5. Cross-thread blocking lock (Thread A holds for 1 second, Thread B blocks)

8.7 Set (Collection)

Chapter Overview: CatBase provides the set[T] collection type, implemented on top of a hash table. Elements are unique and unordered. It supports O(1) add / remove / contains operations, plus the full set of algebraic operations: union / intersection / difference / symmetric difference / subset checks.

8.7.1 What is a Set

A Set is a data structure whose elements are unique and unordered. CatBase's set[T] is built on Zig's std.HashMap and provides:

  • O(1) CRUD: add / remove / contains are all constant-time
  • Set algebra: unionSet (union) / intersection / difference / symmetricDifference
  • Subset checks: isSubset / isSuperset / equals / intersects
  • Automatic deduplication: when a set is created from a list, duplicate elements are silently dropped
  • Interoperable with list: toList converts a set to a list

Comparison with list:

  • list[T]: ordered, allows duplicates — use when order matters or duplicates are allowed
  • set[T]: unordered, unique — use for deduplication, set algebra, and fast membership tests

8.7.2 Type Syntax

CatBase uses set[T] to denote a set whose element type is T, where T must be hashable:


s:set[int]    # a set whose elements are int
s:set[str]    # a set whose elements are str

Supported types: CatBase currently ships with built-in support for set[int] and set[str]. Other element types (such as set[float] or set[bool]) have no top-level constructors yet — the underlying Zig type is already generic, so extending support is straightforward.

8.7.3 Creating a Set

CatBase provides a Python-style set() constructor (recommended), while keeping the legacy top-level functions for backward compatibility:


# === Python-style set() constructor (recommended) ===
# 0-arg: requires LHS type context (compiler infers T from the LHS)
s:set[int] = set()       # empty set[int]
t:set[str] = set()       # empty set[str]

# 1-arg: build from a list (polymorphic dispatch on list element type)
nums:set[int]  = set([1, 2, 3, 2, 1])    # dedup → {1, 2, 3}
words:set[str] = set(["a", "b", "a"])    # dedup → {'a', 'b'}

# === Legacy top-level functions (backward compatible) ===
old:set[int] = set_int()
old:set[int] = set_from_list_int([1, 2, 3, 2, 1])   # legacy 1-arg form
old:set[str] = set_from_list_str(["a", "b", "a"])

set() constructor semantics (Python-compatible):

CatBasePythonSource of element type T
set()set()0-arg has no Python-style context, must be inferred from LHS (strong monomorphism)
set([1,2,3])set([1,2,3])1-arg: T inferred from list element type
set([1,2,3]) assigned to s: set[int]set([1,2,3])1-arg: list element type must match LHS

Why does set() 0-arg require LHS inference?

CatBase's type system is strongly monomorphic: the element type of list[T] / set[T] / dict[K,V] must be known at compile time. set() has no arguments to infer T from, so an LHS type context is required:


# ✅ Correct: LHS provides the set[str] context
s: set[str] = set()

# ❌ Error: T cannot be inferred
# x = set()  # compile error: no LHS type context

# ❌ Error: argument type not supported
# s = set("Tom")   # str is not a list; set() only accepts list[int]/list[str]
# s = set([1,"a"]) # mixed-type list; set() requires a single T in list[X]

Nested scenarios: the LHS context propagates through struct/dict/list literals, so nested constructors can infer their element type layer by layer:


# nested dict (values are sets) — compile-time type inference
d: dict[str, set[int]] = {"a": set(), "b": set([1,2,3])}

# nested struct (fields are sets) — compile-time type inference
struct vst { key: set[str]; value: set[str] }
v: vst = vst { key: set(), value: set() }                   # set() infers from field types

Full real-world example: struct literal with nested set (social network scenario):


struct Profile {
    friends: set[str]
    tags:    set[str]
}

def main() {
    alice: Profile = Profile {
        friends: set(["bob", "carol", "dave"]),
        tags:    set(["rust", "zig", "python"])
    }
    print(alice.friends)        # {'carol', 'dave', 'bob'}
    print(alice.tags)           # {'zig', 'python', 'rust'}

    # cross-field set operations (social graph)
    bob: Profile = Profile {
        friends: set(["alice", "carol", "eve"]),
        tags:    set(["rust", "go"])
    }
    common: set[str] = alice.friends.intersection(bob.friends)
    print("mutual friends:", common)    # {'carol'}
}

Note: in the current version, print of nested set types still uses Zig's default HashMap debug format (e.g. {"1":"void", ...}) rather than the CatBase-style set output. The compile-time type inference for set() 0-arg / 1-arg is fully working; only the nested print formatting needs further polish.

FunctionDescriptionRecommendation
set() 0-argT inferred from LHS, returns empty set⭐⭐⭐ Recommended
set([...]) 1-argBuild from list, dedup automatically⭐⭐⭐ Recommended
set_int() / set_str()Create empty set[int] / set[str]Deprecated
set_from_list_int(list) / set_from_list_str(list)Legacy 1-arg formDeprecated

Note: set instances are allocated from the global allocator and live for the entire program. You do not need to call deinit.

8.7.4 Basic Operations


s:set[int] = set_int()

# add elements (duplicates are silently ignored)
s.add(10)
s.add(20)
s.add(10)  # already present, no change
# s = {10, 20}

# check whether an element is present
print(s.contains(10))  # true
print(s.contains(99))  # false

# get the number of elements
print(s.len())  # 2

# check whether the set is empty
print(s.isEmpty())  # false

# remove an element (returns whether it was actually removed)
removed:bool = s.remove(10)  # true
removed2:bool = s.remove(99)  # false

# clear all elements (keeps the underlying capacity)
s.clear()
print(s.isEmpty())  # true

8.7.5 Set Algebra

CatBase provides four basic set operations. All of them return a new set; they never modify the operands.


A:set[int] = set_from_list_int([1, 2, 3, 4])
B:set[int] = set_from_list_int([3, 4, 5, 6])

# union: A ∪ B = {1, 2, 3, 4, 5, 6}
u:set[int] = A.unionSet(B)
# print(u) => {1, 2, 3, 4, 5, 6}

# intersection: A ∩ B = {3, 4}
inter:set[int] = A.intersection(B)
# print(inter) => {3, 4}

# difference: A - B = {1, 2} (elements in A but not in B)
diff_ab:set[int] = A.difference(B)
# print(diff_ab) => {1, 2}

# symmetric difference: (A ∪ B) - (A ∩ B) = {1, 2, 5, 6}
sym:set[int] = A.symmetricDifference(B)
# print(sym) => {1, 2, 5, 6}

# the original sets are left untouched
# print(A) => {1, 2, 3, 4}
# print(B) => {3, 4, 5, 6}

⚠️ Important: the union method is called unionSet, not union

This is because union is a reserved keyword in Zig (it is used to declare tagged union types). Naming the method union would clash with that keyword and fail to compile. CatBase intentionally uses unionSet to avoid the clash.

The other set operations (intersection / difference / symmetricDifference) do not collide with Zig keywords and follow standard mathematical terminology.

8.7.6 Subset, Superset, and Intersection Checks


S:set[int] = set_from_list_int([1, 2])
T:set[int] = set_from_list_int([1, 2, 3])
U:set[int] = set_from_list_int([1, 2, 3])

# subset: self ⊆ other (every element of self is in other)
print(S.isSubset(T))      # true   S is a subset of T
print(T.isSubset(U))      # true   equal sets are mutual subsets

# superset: self ⊇ other
print(T.isSuperset(S))    # true   T is a superset of S

# equality: exactly the same elements (order-independent)
print(T.equals(U))        # true
print(S.equals(T))        # false (different sizes)

# whether two sets share any element
P:set[int] = set_from_list_int([1, 2, 3])
Q:set[int] = set_from_list_int([4, 5, 6])
print(P.intersects(Q))    # false  no shared elements
print(P.intersects(T))    # true   shared elements

8.7.7 Converting Between Sets and Lists


# set -> list (order is NOT guaranteed)
nums:set[int] = set_from_list_int([3, 1, 2])
lst:list[int] = nums.toList()
# print(lst) => [1, 2, 3] (order depends on the hash function)

# list -> set (duplicates are dropped)
raw:list[int] = [1, 2, 2, 3, 3, 3]
unique:set[int] = set_from_list_int(raw)
# print(unique) => {1, 2, 3}

8.7.8 Important Notes

  1. Uniqueness

    Sets automatically deduplicate. Calling add with an element that is already present has no effect.

  2. Order is not preserved

    Sets are unordered. The list produced by toList follows the hash function's internal order, do not depend on any particular ordering.

  3. No index access

    Sets do not support s[0]-style indexing, because they are unordered. If you need indexed access, convert to a list with toList first.

  4. No direct iteration (current version)

    CatBase does not yet provide for x in s syntax for sets. To iterate, use toList:

    
    s:set[int] = set_from_list_int([1, 2, 3])
    for x in s.toList() {
        print(x)
    }
    
  5. Printing a set directly has limitations (to be improved)

    Calling print(s) prints a representation of the form {...}, but using an f-string like print(f"{s}") only prints the literal {s} instead of the actual contents. Prefer print(s) over print(f"...{s}...").

  6. Element type is fixed

    set[int] and set[str] are distinct types and cannot be mixed:

    
    s:set[int] = set_int()
    s.add("hello")  # ✗ error: set[int] cannot hold a str
    t:set[str] = s  # ✗ error: type mismatch
    
  7. Union method is named unionSet, not union

    Because union is a reserved keyword in Zig.

8.7.9 Full Method Reference

Method Parameter Return type Description
add(value) T bool Insert an element. Returns true if newly inserted, false if already present.
remove(value) T bool Remove an element. Returns true if it was actually removed.
contains(value) T bool Check whether the element is in the set.
len() int Get the number of elements.
isEmpty() bool Check whether the set is empty.
clear() void Remove all elements (keeps the underlying capacity).
unionSet(other) set[T] set[T] Union A ∪ B.
intersection(other) set[T] set[T] Intersection A ∩ B.
difference(other) set[T] set[T] Difference A - B.
symmetricDifference(other) set[T] set[T] Symmetric difference (A ∪ B) - (A ∩ B).
isSubset(other) set[T] bool self ⊆ other.
isSuperset(other) set[T] bool self ⊇ other.
equals(other) set[T] bool True if both sets have the same elements.
intersects(other) set[T] bool True if the two sets share at least one element.
toList() list[T] Convert to a list (order not guaranteed).

8.7.10 Complete Example


# deduplicate a list
raw:list[int] = [5, 3, 5, 1, 3, 2, 1, 4, 5]
unique:set[int] = set_from_list_int(raw)
print("unique element count:", unique.len())  # 5

# find elements unique to each of two sets
team_a:set[str] = set_str()
team_a.add("Alice")
team_a.add("Bob")
team_a.add("Carol")

team_b:set[str] = set_str()
team_b.add("Carol")
team_b.add("Dave")
team_b.add("Eve")

only_a:set[str] = team_a.difference(team_b)         # {'Alice', 'Bob'}
only_b:set[str] = team_b.difference(team_a)         # {'Dave', 'Eve'}
both:set[str]   = team_a.intersection(team_b)         # {'Carol'}

# combined: deduplication + set algebra
nums:set[int] = set_from_list_int([1, 2, 3, 4, 5])
evens:set[int] = set_from_list_int([2, 4, 6, 8])
overlap:set[int] = nums.intersection(evens)  # {2, 4}

A complete test driver is available at examples/test_set.cat, covering 12 scenarios:

  1. Basic set[int] operations (add/remove/contains/len/clear/isEmpty)
  2. Basic set[str] operations
  3. Build from a list (set_from_list_int / set_from_list_str)
  4. Union (unionSet)
  5. Intersection (intersection)
  6. Difference (difference)
  7. Symmetric difference (symmetricDifference)
  8. Subset / superset checks (isSubset / isSuperset / equals)
  9. Intersection check (intersects)
  10. Convert to list (toList)
  11. Composite example: list deduplication
  12. Composite example: find elements unique to each of two sets

8.7.11 Unified hash() Function

CatBase provides a unified hash(x) function for 64-bit hashing of containers (list / set / dict). The compiler dispatches to one of 35 monomorphic runtime implementations based on the argument type — no need to call different hash APIs manually.

Function signature:


hash(x: list[T] | set[T] | dict[K, V] | list[list[T]] | list[set[T]] | list[dict[K, V]]) : int

T, K, and V currently support the three leaf types int / str / float.

Core rules:

ContainerOrder sensitivityNotes
list[X]Order-sensitiveCombine in insertion order; [1,2,3][3,2,1]
set[X]Order-insensitiveInternal sort then combine; {1,2,3} = {3,2,1}
dict[K,V]Order-insensitiveSort by key then combine; {"a":1,"b":2} = {"b":2,"a":1}
list[list[X]]Outer + inner list both sensitiveEvery list layer combines by position
list[set[X]]Outer sensitive + inner set insensitiveNested sets auto-sort internally
list[dict[K,V]]Outer sensitive + inner dict insensitiveNested dicts sort by key internally

Cross-type non-collision: every (outer container × inner element) combination has its own 64-bit type tag, so:

  • hash([1, 2, 3])hash(["1", "2", "3"])hash([1.0, 2.0, 3.0])
  • hash([1, 2, 3])hash(set([1, 2, 3]))hash({"a": 1})
  • hash([[1,2]])hash([set([1,2])])hash([{"a":1}])

Return type is int in CatBase (underlying u64, safely cast to i64 via @intCast, bit pattern unchanged).

Example:


def main() {
    # 1. list is order-sensitive
    a: int = hash([1, 2, 3])
    b: int = hash([3, 2, 1])
    print("a != b (order-sensitive):", a != b)   # true

    # 2. set / dict are order-insensitive
    s1: int = hash(set([1, 2, 3]))
    s2: int = hash(set([3, 2, 1]))
    print("s1 == s2 (order-insensitive):", s1 == s2)  # true

    d1: int = hash({"a": 1, "b": 2})
    d2: int = hash({"b": 2, "a": 1})
    print("d1 == d2 (order-insensitive):", d1 == d2)  # true

    # 3. Cross-type non-collision
    print(hash([1, 2, 3]) != hash(["1", "2", "3"]))   # true
    print(hash([1, 2, 3]) != hash([1.0, 2.0, 3.0]))  # true

    # 4. Nesting: outer sensitive + inner set/dict insensitive
    x: int = hash([set([1, 2]), set([3, 4])])
    y: int = hash([set([4, 3]), set([2, 1])])   # each inner set reordered
    z: int = hash([set([3, 4]), set([1, 2])])   # outer list reordered
    print("x == y (inner set insensitive):", x == y)   # true
    print("x != z (outer list sensitive):", x != z)   # true
}

Use cases:

  • Cache keys: hash a list / set / dict into a 64-bit int usable as a Dict key (CatBase's Dict[K,V] requires hashable keys; int is naturally hashable).
  • Dedup fingerprinting: stamp a config / state snapshot for cross-process or cross-language comparison.
  • Fast comparison: a single O(n) hash pass beats depth-first equals traversal in order-sensitive scenarios.
  • Hash table / integrity check: map any nested container to a stable 64-bit digest.

Implementation notes:

  • Compile-time dispatch: hash([1,2,3]) emits runtime.hashListInt(...); hash(set([1,2,3])) emits runtime.hashSetInt(...); and so on.
  • Combine algorithm is boost-style h ^= b + 0x9E3779B97F4A7C15 + (h<<6) + (h>>2) with the golden ratio constant.
  • Up to 1024 elements per container (overflow protection, extra elements are silently truncated; only safe for bounded data — split large sets via set_to_list first).
  • Does not support deeper nesting such as set[set[X]] / dict[set[X]] / set[dict[K,V]] (only the list[nested] layer is supported); for deeper nesting, manually apply hash(hash(x)) to fold layers.

Related sections: 8.7 set container / 8.6 list container / 8.8 dict container.


9. Network Programming

Chapter Overview: File operations let us handle local data; this chapter introduces network programming. CatBase has built-in powerful networking capabilities, supporting common protocols such as TCP, UDP, and HTTP, enabling you to easily develop network applications.

9.1 TCP Sockets

CatBase provides a unified TCPSocket type that can be used to create TCP clients or TCP servers.

Creating a TCP Socket

tcpsocket()

Creates a new TCP socket.


# TCP client
sock:TCPSocket = tcpsocket()
sock.connect("example.com", 80)

TCPSocket Methods

Method Description
sock.connect(host, port[, timeout]) Connect to a TCP server (client mode), with optional timeout parameter
sock.bind(host, port) Bind address and port (server mode)
sock.listen(backlog) Start listening for connections (server mode)
sock.accept() Accept a connection, returns TCPClient (server mode)
sock.write(data) Send data
sock.read(size) Receive up to size bytes of data
sock.close() Close socket

TCP Client Example


def main(args:list[str]) {
    client:TCPSocket = tcpsocket()
    client.connect("example.com", 80)

    client.write("GET / HTTP/1.0\r\n\r\n")
    response:str = client.read(4096)
    print("Response length: ", len(response), "\n")
    client.close()
}

9.2 TCP Server


def main(args:list[str]) {
    server:TCPSocket = tcpsocket()
    server.bind("0.0.0.0", 8080)
    server.listen(128)
    print("Server listening on port 8080\n")

    conn:TCPClient = server.accept()
    print("Client connected\n")

    data:str = conn.read(1024)
    print("Received: ", data, "\n")

    conn.write("HTTP/1.0 200 OK\r\n\r\nHello!")
    conn.close()
    server.close()
}

9.3 UDP Sockets

CatBase provides the UDPSocket type for UDP communication.

Creating a UDP Socket

udpsocket()

Creates a new UDP socket.


udp:UDPSocket = udpsocket()

UDPSocket Methods

Method Description
udp.bind(host, port) Bind address and port
udp.sendto(data, host, port) Send data to specified address
udp.recvfrom(size[, timeout]) Receive data, returns string, with optional timeout parameter
udp.close() Close socket

UDP Client Example


def main(args:list[str]) {
    udp:UDPSocket = udpsocket()
    udp.sendto("Hello from UDP!", "127.0.0.1", 9999)
    udp.close()
}

UDP Server Example


def main(args:list[str]) {
    udp:UDPSocket = udpsocket()
    udp.bind("0.0.0.0", 9999)
    print("UDP server listening on port 9999\n")

    try {
        # Use receive method with timeout, 10 seconds timeout
        data:str = udp.recvfrom(1024, 10)
        print("Received: ", data, "\n")
        
        # Send response to client
        udp.sendto("Received: " + data, "127.0.0.1", 9998)
    }
    except err {
        print("Error: ", err, "\n")
    }

    udp.close()
}

9.4 Comparison with Python socket

CatBase Python
tcpsocket() socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(host, port) sock.connect((host, port))
sock.bind(host, port) sock.bind((host, port))
sock.listen(backlog) sock.listen(backlog)
sock.accept() conn, addr = sock.accept()
sock.write(data) sock.send(data)
sock.read(size) sock.recv(size)
udpsocket() socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
udp.sendto(data, host, port) sock.sendto(data, (host, port))
udp.recvfrom(size) data, addr = sock.recvfrom(size)

9.5 HTTP Requests

http_get

http_get(url:str, timeout:int) : str - Send HTTP GET request

The http_get function is used to send HTTP GET requests and supports setting a timeout.

Parameter description:

Parameter Type Required Default Description
url str Yes - The request URL address
timeout int No 60 Timeout in seconds

Return value:

  • Success: Returns HTTP response content (str type)
  • Failure: Throws an error, such as error.Timeout (connection timeout), error.HostUnreachable (host unreachable)

Basic usage:


def main(args:list[str]) {
    # Simple GET request (using default timeout of 60 seconds)
    response:str = http_get("http://example.com")
    print("Response: ", response, "\n")
}

GET request with timeout parameter:


def main(args:list[str]) {
    # Set 10 seconds timeout
    response:str = http_get("http://example.com", timeout=10)
    print("Response: ", response, "\n")
}

Using try-except to catch errors:


def main(args:list[str]) {
    try {
        # Try GET request, 3 seconds timeout
        response:str = http_get("http://192.168.254.254:9999/", timeout=3)
        print("Response: ", response, "\n")
    }
    except err {
        print("HTTP GET failed with error: ", err, "\n")
    }
}

Run result (timeout case):


HTTP GET failed with error: error.Timeout

GET request with query parameters:

Use ? followed by query parameters in the URL, with multiple parameters separated by &.


def main(args:list[str]) {
    # GET request with query parameters
    # Query parameters are appended directly to the URL
    url:str = "http://httpbin.org/get?username=admin&password=123456"
    response:str = http_get(url, timeout=10)
    print("Response: ", response, "\n")
}

Run result:


Response: {
  "args": {
    "password": "123456", 
    "username": "admin"
  }, 
  "headers": {
    "Host": "httpbin.org"
  }, 
  "origin": "xxx.xxx.xxx.xxx", 
  "url": "http://httpbin.org/get?username=admin&password=123456"
}

http_post

http_post(url:str, headers:str, json:dict, data:str, timeout:int, stream:bool) : Response - Send HTTP POST request

The http_post function supports the following parameters (all can be passed as keyword arguments):

Parameter Type Required Default Description
url str Yes - The request URL address
headers str No "" Request headers
json dict No {} JSON data (automatically sets Content-Type)
data str No "" Request body data
timeout int No 60 Timeout (seconds)
stream bool No False Whether to use streaming response
Form Data Submission

Submit form data using the application/x-www-form-urlencoded format:


def main(args:list[str]) {
    # Form data submission (using data keyword argument)
    response:Response = http_post("http://httpbin.org/post", data="username=admin&password=123456")
    print("Response: ", response, "\n")
}

Run result:


Response: {
  "args": {}, 
  "data": "", 
  "files": {}, 
  "form": {
    "password": "123456", 
    "username": "admin"
  }, 
  "headers": {
    "Content-Length": "27", 
    "Content-Type": "application/x-www-form-urlencoded", 
    "Host": "httpbin.org"
  }, 
  "json": null, 
  "origin": "xxx.xxx.xxx.xxx", 
  "url": "http://httpbin.org/post"
}

JSON Data Submission

To submit data in JSON format, you need to manually construct the JSON string:


def main(args:list[str]) {
    # Construct JSON data
    # Format: {"key": "value"}
    json_data:str = "{\"username\": \"admin\", \"password\": \"123456\", \"age\": 25}"
    
    response:Response = http_post("http://httpbin.org/post", data=json_data)
    print("Response: ", response, "\n")
}

Run result:


Response: {
  "args": {}, 
  "data": "{\"username\": \"admin\", \"password\": \"123456\", \"age\": 25}", 
  "files": {}, 
  "form": {}, 
  "headers": {
    "Content-Length": "53", 
    "Content-Type": "application/x-www-form-urlencoded", 
    "Host": "httpbin.org"
  }, 
  "json": {
    "age": 25, 
    "password": "123456", 
    "username": "admin"
  }, 
  "origin": "xxx.xxx.xxx.xxx", 
  "url": "http://httpbin.org/post"
}

Submitting Complex JSON Data

def main(args:list[str]) {
    # Submit JSON containing an array
    json_data:str = "{\"users\": [{\"name\": \"Alice\", \"age\": 30}, {\"name\": \"Bob\", \"age\": 25}], \"status\": \"active\"}"
    
    response:Response = http_post("http://httpbin.org/post", data=json_data)
    print("Response: ", response, "\n")
}

Run result:


Response: {
  "args": {}, 
  "data": "{\"users\": [{\"name\": \"Alice\", \"age\": 30}, {\"name\": \"Bob\", \"age\": 25}], \"status\": \"active\"}", 
  "files": {}, 
  "form": {}, 
  "headers": {
    "Content-Length": "87", 
    "Content-Type": "application/x-www-form-urlencoded", 
    "Host": "httpbin.org"
  }, 
  "json": {
    "status": "active", 
    "users": [
      {
        "age": 30, 
        "name": "Alice"
      }, 
      {
        "age": 25, 
        "name": "Bob"
      }
    ]
  }, 
  "origin": "xxx.xxx.xxx.xxx", 
  "url": "http://httpbin.org/post"
}

Note: http_post uses application/x-www-form-urlencoded Content-Type by default. If you need to submit pure JSON data, you need to parse the data field on the server side.

9.5.1 OpenAI API Calls

The OpenAI API allows developers to access powerful large language models (LLMs), such as GPT-4, GPT-3.5, etc. CatBase can communicate with the OpenAI API through the http_post function to implement natural language processing, intelligent conversations, and more.

Basic Chat Request


def main(args:list[str]) {
    # OpenAI API configuration
    api_key:str = "your-openai-api-key"
    api_url:str = "https://api.openai.com/v1/chat/completions"

    # Construct request JSON
    # messages array contains conversation history
    # role: system, user, assistant
    # content: message content
    request_json:str = "{\"model\": \"gpt-3.5-turbo\", \"messages\": [{\"role\": \"system\", \"content\": \"You are a helpful assistant.\"}, {\"role\": \"user\", \"content\": \"Hello, who are you?\"}]}"

    # Send POST request
    response:Response = http_post(api_url, data=request_json)
    print("Response: ", response, "\n")
}

Sending Conversation Request and Parsing Response


def main(args:list[str]) {
    # OpenAI API configuration
    api_key:str = "your-openai-api-key"
    api_url:str = "https://api.openai.com/v1/chat/completions"

    # Construct conversation request
    # system: defines assistant behavior
    # user: user input
    request_json:str = "{\"model\": \"gpt-3.5-turbo\", \"messages\": [{\"role\": \"system\", \"content\": \"You are a professional Python programming assistant.\"}, {\"role\": \"user\", \"content\": \"Please write a quicksort algorithm in Python.\"}]}"

    # Send request
    response:Response = http_post(api_url, data=request_json)
    print("API Response:\n")
    print(response)
    print("\n")

    # Response format:
    # {
    #   "id": "chatcmpl-...",
    #   "choices": [{
    #     "message": {
    #       "role": "assistant",
    #       "content": "..."
    #     }
    #   }]
    # }
}

Multi-turn Conversation


def main(args:list[str]) {
    api_url:str = "https://api.openai.com/v1/chat/completions"

    # Initial conversation history
    # You can append previous conversations to this array to implement multi-turn conversation
    messages:str = "[{\"role\": \"system\", \"content\": \"You are a helpful assistant.\"}, {\"role\": \"user\", \"content\": \"What is artificial intelligence?\"}]"

    # Send first round of conversation
    request_json:str = "{\"model\": \"gpt-3.5-turbo\", \"messages\": " + messages + "}"
    response:Response = http_post(api_url, data=request_json)

    print("First response:\n")
    print(response)
    print("\n")

    # To continue the conversation, you can parse the response and append to messages
    # Then send a new request...

    # Second round example (need to manually append previous conversation)
    messages:str = "[{\"role\": \"system\", \"content\": \"You are a helpful assistant.\"}, {\"role\": \"user\", \"content\": \"What is artificial intelligence?\"}, {\"role\": \"assistant\", \"content\": \"Artificial intelligence is...\"}, {\"role\": \"user\", \"content\": \"What are its applications?\"}]"

    request_json = "{\"model\": \"gpt-3.5-turbo\", \"messages\": " + messages + "}"
    response = http_post(api_url, data=request_json)

    print("Second response:\n")
    print(response)
    print("\n")
}

Setting Generation Parameters

The OpenAI API supports multiple generation parameters to control the quality and diversity of output:


def main(args:list[str]) {
    api_url:str = "https://api.openai.com/v1/chat/completions"

    # Construct request with parameters
    # temperature: 0.0-2.0, controls randomness. Lower values make output more deterministic, higher values make output more random
    # max_tokens: maximum number of tokens to generate
    # top_p: nucleus sampling parameter
    request_json:str = "{\"model\": \"gpt-3.5-turbo\", \"messages\": [{\"role\": \"user\", \"content\": \"Write a joke about cats.\"}], \"temperature\": 0.8, \"max_tokens\": 100}"

    response:Response = http_post(api_url, data=request_json)
    print("Response with parameters:\n")
    print(response)
    print("\n")
}

Streaming Response

Streaming responses allow real-time display of LLM output without waiting for the complete response. It is especially useful for long text generation.


def main(args:list[str]) {
    api_url:str = "https://api.openai.com/v1/chat/completions"

    # Enable stream: true to get streaming response
    request_json:str = "{\"model\": \"gpt-3.5-turbo\", \"messages\": [{\"role\": \"user\", \"content\": \"Explain quantum computing in three sentences.\"}], \"stream\": true}"

    response:Response = http_post(api_url, data=request_json)
    print("Streaming response:\n")
    print(response)
    print("\n")

    # Note: In actual use, streaming responses return data in SSE format
    # You need to parse lines starting with data: yourself
}

Streaming Response Data Parsing

Streaming responses return data in SSE (Server-Sent Events) format, with each line starting with data:. Here is an example of parsing a streaming response:


def main(args:list[str]) {
    api_url:str = "https://api.openai.com/v1/chat/completions"

    # Enable streaming response
    request_json:str = "{\"model\": \"gpt-3.5-turbo\", \"messages\": [{\"role\": \"user\", \"content\": \"Explain what recursion is.\"}], \"stream\": true}"

    response:Response = http_post(api_url, data=request_json)

    # Print raw response
    print("Raw streaming response:\n")
    print(response)
    print("\n")

    # Streaming response format example:
    # data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","created":1234567890,"model":"gpt-3.5-turbo","choices":[{"index":0,"delta":{"content":"Recursion"},"finish_reason":null}]}
    #
    # data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","created":1234567890,"model":"gpt-3.5-turbo","choices":[{"index":0,"delta":{"content":" is a"},"finish_reason":null}]}
    #
    # data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","created":1234567890,"model":"gpt-3.5-turbo","choices":[{"index":0,"delta":{"content":" kind"},"finish_reason":null}]}
    #
    # data: [DONE]

    # Extract content from each delta
    # Note: Actual parsing requires string processing functions, here only shows the concept
    print("Streaming complete!\n")
}

Complete Streaming Chat Example


def main(args:list[str]) {
    api_url:str = "https://api.openai.com/v1/chat/completions"

    # Construct request
    request_json:str = "{\"model\": \"gpt-3.5-turbo\", \"messages\": [{\"role\": \"user\", \"content\": \"Write me a five-line poem about programming.\"}], \"stream\": true}"

    print("Requesting streaming response...\n")
    print("Response: \n")

    # Send streaming request
    response:Response = http_post(api_url, data=request_json)

    # Print complete response
    print(response)
    print("\n")

    # SSE streaming response format description:
    # 1. Each data chunk starts with "data: "
    # 2. Each chunk is a JSON object containing a delta field
    # 3. delta.content contains incremental text
    # 4. Ends with "data: [DONE]"

    # Simulated parsing process (actually requires string function support)
    # chunks:list = split(response, "data:")
    # for chunk in chunks {
    #     if starts_with(chunk, "[DONE]") {
    #         break
    #     }
    #     # Parse JSON to extract content
    #     content:str = extract_json_field(chunk, "content")
    #     print(content)
    # }

    print("\nStreaming response ended.\n")
}

Application Scenarios for Streaming Response

Scenario Description
Real-time typing effect Simulates typewriter effect, enhancing user experience
Long text generation No need to wait for complete response, display while generating
Interactive conversation Users can interrupt or adjust during generation
Code completion Display code completion suggestions in real time

Complete Example: Intelligent Assistant


def main(args:list[str]) {
    api_url:str = "https://api.openai.com/v1/chat/completions"

    # System prompt, defines assistant role
    system_prompt:str = "You are a professional, friendly programming assistant named CatBot. You excel at explaining programming concepts and helping write code."

    # User's first question
    user_question:str = "Please explain what variables are and why we need to declare variable types?"

    # Construct request
    request_json:str = "{\"model\": \"gpt-3.5-turbo\", \"messages\": [{\"role\": \"system\", \"content\": \"" + system_prompt + "\"}, {\"role\": \"user\", \"content\": \"" + user_question + "\"}], \"temperature\": 0.7, \"max_tokens\": 500}"

    print("Sending request to OpenAI API...\n")

    # Send request
    response:Response = http_post(api_url, data=request_json)

    print("Response received!\n")
    print("=" * 50)
    print(response)
    print("=" * 50)
    print("\n")

    # Tip: In actual applications, you need to parse the JSON response to extract the content field
    # Response format: {"choices": [{"message": {"content": "..."}}]}
}

Other OpenAI API Endpoints

OpenAI also provides other API endpoints that can be called in a similar way:

Endpoint Description URL Format
Chat Multi-turn conversation https://api.openai.com/v1/chat/completions
Completions Text completion https://api.openai.com/v1/completions
Embeddings Text embeddings https://api.openai.com/v1/embeddings
Images Image generation https://api.openai.com/v1/images/generations

Text completion example:


def main(args:list[str]) {
    api_url:str = "https://api.openai.com/v1/completions"

    # Construct completion request
    request_json:str = "{\"model\": \"text-davinci-003\", \"prompt\": \"Once upon a time there was a mountain,\", \"max_tokens\": 50, \"temperature\": 0.7}"

    response:Response = http_post(api_url, data=request_json)
    print("Completion response:\n")
    print(response)
    print("\n")
}

http_post (Recommended)

http_post(url:str, headers:str, json:dict[str,any], data:str, timeout:int, stream:bool) : Response - Send HTTP POST request (recommended)

The http_post function is an advanced HTTP POST function similar to Python's requests.post(), supporting multiple parameters:

  • url (str): Request URL (required)
  • headers (str): Request headers, multiple headers separated by & (optional, default empty string)
  • json (dict\[str,any]): JSON data, accepts dict type (optional)
  • data (str): Form data (optional, ignored when json has a value)
  • timeout (int): Timeout in seconds, default 60 (optional)
  • stream (bool): Whether to use streaming mode, default False (optional)

Parameter priority: When both json and data parameters have values, the json parameter takes precedence.

Automatic Content-Type detection: The http_post function automatically sets Content-Type based on the parameter type passed in:

  • If the json parameter is passed, automatically set to application/json and serialize the dictionary to JSON
  • If the data parameter is passed, automatically set to application/x-www-form-urlencoded

Return value: Returns a Response object containing the following methods:

  • raise_for_status(): Check response status code, triggers exception if not 200-299
  • iter_lines(): Iterate to read streaming response content (only available when stream=True)
Basic Usage

def main(args:list[str]) {
    url:str = "http://httpbin.org/post"
    
    # Use json parameter to submit JSON data
    json_data:dict[str,any] = {"name": "Tom", "age": 20}
    response:Response = http_post(url=url, json=json_data)
    print("Response: ", response, "\n")
}
With Request Headers

def main(args:list[str]) {
    url:str = "http://httpbin.org/post"
    headers:str = "Authorization: Bearer abc123"
    json_data:dict[str,any] = {"username": "admin", "password": "123456"}
    
    response:Response = http_post(url=url, headers=headers, json=json_data)
    print("Response: ", response, "\n")
}
Submitting Form Data with data Parameter

def main(args:list[str]) {
    url:str = "http://httpbin.org/post"
    
    # Use data parameter to submit form data
    form_data:str = "username=admin&password=123456"
    response:Response = http_post(url=url, data=form_data)
    print("Response: ", response, "\n")
}
Setting Timeout

def main(args:list[str]) {
    url:str = "http://httpbin.org/post"
    json_data:dict[str,any] = {"name": "test"}
    
    # Set timeout to 30 seconds
    response:Response = http_post(url=url, json=json_data, timeout=30)
    print("Response: ", response, "\n")
}
OpenAI API Call

def main(args:list[str]) {
    api_key:str = "your-openai-api-key"
    api_url:str = "https://api.openai.com/v1/chat/completions"
    
    # Construct request data
    request_data:dict[str,any] = {
        "model": "gpt-3.5-turbo",
        "messages": [
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": "Hello, who are you?"}
        ],
        "temperature": 0.7,
        "max_tokens": 100
    }
    
    headers:str = "Authorization: Bearer " + api_key
    response:Response = http_post(url=api_url, headers=headers, json=request_data)
    print("Response: ", response, "\n")
}
Streaming Response (for vLLM, OpenAI, etc. APIs)

def main(args:list[str]) {
    # vLLM or OpenAI compatible API call
    base_url:str = "http://localhost:19090/v1"
    model_path:str = "qwen"
    url:str = base_url + "/chat/completions"
    
    # Construct request data
    data:dict[str,any] = {
        "model": model_path,
        "messages": [
            {"role": "system", "content": "Answer the question directly, without any thinking process."},
            {"role": "user", "content": "Hello"}
        ],
        "temperature": 0.1,
        "max_tokens": 1000,
        "stream": True
    }
    
    # Send request and enable streaming mode
    response:Response = http_post(url, json=data, stream=True, timeout=60)
    
    # Check response status
    response.raise_for_status()
    
    # Iterate to read streaming response
    for line in response.iter_lines() {
        print(line)
    }
}

HTTP Response Object

When using the stream=True parameter, http_post returns a Response object containing the following methods:

raise_for_status

response.raise_for_status() - Check HTTP response status

If the response status code is not in the 200-299 range, a panic exception is triggered. This is consistent with the behavior of Python's requests.Response.raise_for_status() method.


def main(args:list[str]) {
    response:Response = http_post(url, json=data, stream=True)
    
    # Check status code, non-200-299 triggers exception
    response.raise_for_status()
    
    # Continue processing response...
}
iter_lines

response.iter_lines() - Iterate to read streaming response lines

This method returns an iterator that returns one line of the response per call. Returns None when the end of the response is reached.


def main(args:list[str]) {
    response:Response = http_post(url, json=data, stream=True)
    
    # Method 1: Iterate using for loop (recommended)
    for line in response.iter_lines() {
        print(line)
    }
    
    # Method 2: Using while loop
    while True {
        line:str = response.iter_lines()
        if !line {
            break
        }
        print(line)
    }
}

json_dumps

json_dumps(data:dict[str,any]) : str - Convert dictionary to JSON string

The json_dumps function is used to serialize CatBase's dictionary type (dict\[str,any]) to a JSON string, similar to Python's json.dumps() function.

Basic Usage

def main(args:list[str]) {
    data:dict[str,any] = {"name": "Tom", "age": 20}
    json_str:str = json_dumps(data)
    print("JSON: ", json_str, "\n")
}

Run result:


JSON:  {"name":"Tom","age":20}

Mixed Type Data

def main(args:list[str]) {
    # Supports multiple data types
    data:dict[str,any] = {
        "name": "Alice",
        "age": 25,
        "score": 98.5,
        "active": True,
        "tags": ["developer", "programmer"],
        "info": {"city": "Beijing", "country": "China"}
    }
    
    json_str:str = json_dumps(data)
    print("JSON: ", json_str, "\n")
}

Run result:


JSON:  {"name":"Alice","age":25,"score":98.5,"active":true,"tags":["developer","programmer"],"info":{"city":"Beijing","country":"China"}}

Nested Lists

def main(args:list[str]) {
    # Dictionary containing a list
    items:list[str] = ["apple", "banana", "orange"]
    
    data:dict[str,any] = {
        "product": "fruit",
        "items": items,
        "count": 3
    }
    
    json_str:str = json_dumps(data)
    print("JSON: ", json_str, "\n")
}
List of Dictionaries

def main(args:list[str]) {
    # List containing dictionaries
    users:list[dict[str,str]] = [
        {"name": "Tom", "age": "20"},
        {"name": "Jerry", "age": "25"}
    ]
    
    data:dict[str,any] = {
        "users": users,
        "total": 2
    }
    
    json_str:str = json_dumps(data)
    print("JSON: ", json_str, "\n")
}
json_loads

json_loads(json_str:str) : dict[str,any] - Parse JSON string to dictionary

The json_loads function is used to parse a JSON string into CatBase's dictionary type (dict\[str,any]), the opposite of Python's json.loads() function.

Basic Usage

def main(args:list[str]) {
    json_str:str = "{\"name\": \"Alice\", \"age\": 30}"
    
    data:dict[str,any] = json_loads(json_str)
    print("Parsed: ", data, "\n")
}

Run result:


Parsed: {"age":30,"name":"Alice"}

Parsing Nested JSON

def main(args:list[str]) {
    json_str:str = "{\"name\": \"Bob\", \"info\": {\"city\": \"Beijing\", \"country\": \"China\"}, \"scores\": [90, 85, 92]}"
    
    data:dict[str,any] = json_loads(json_str)
    print("Parsed: ", data, "\n")
}

Run result:


Parsed: {"info":{"city":"Beijing","country":"China"},"name":"Bob","scores":[90,85,92]}

Using with http_post

def main(args:list[str]) {
    url:str = "http://httpbin.org/post"
    
    # Construct JSON data
    request_data:dict[str,any] = {
        "model": "gpt-3.5-turbo",
        "messages": [
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": "Hello!"}
        ],
        "temperature": 0.7
    }
    
    # Convert to JSON string (optional, http_post can also accept dict directly)
    json_str:str = json_dumps(request_data)
    print("Sending JSON: ", json_str, "\n")
    
    # Send request (json parameter accepts dict directly)
    response:Response = http_post(url=url, json=request_data)
    print("Response: ", response, "\n")
}
Dictionary .get() Method

The dictionary type supports the .get() method for safely retrieving values from a dictionary. When the key does not exist, it returns a default value, avoiding program crashes.


def main(args:list[str]) {
    # Parse JSON string
    json_str:str = "{\"name\": \"Alice\", \"age\": 30}"
    data:dict = json_loads(json_str)
    
    # Use .get() method to get value
    name:any = data.get("name", "")
    city:any = data.get("city", "Unknown")
    
    print("Name: ", str(name), "\n")
    print("City: ", str(city), "\n")
}

Run result:


Name: Alice
City: Unknown

Syntax of .get() Method

dict.get(key, default)
  • key (str): The key name to retrieve
  • default: The default value returned when the key does not exist, must be specified

Important Features:

  • .get() method must be passed a second parameter (default value), otherwise a compilation error occurs
  • The return value type is determined by the type of the second parameter:
    • .get("name", "") returns str type
    • .get("count", 0) returns int type
    • .get("ratio", 0.5) returns float type
    • .get("enabled", True) returns bool type
    • .get("data", {}) returns dict[str, any] type
    • .get("items", []) returns list[any] type

Example:


def main(args:list[str]) {
    # Parse JSON string
    json_str:str = "{\"name\": \"Alice\", \"age\": 30, \"scores\": [95, 87, 92]}"
    data:dict = json_loads(json_str)
    
    # Infer return type from default value type
    name:str = data.get("name", "")       # Returns str type
    age:int = data.get("age", 0)           # Returns int type
    city:str = data.get("city", "Unknown") # Returns str type
    scores:list = data.get("scores", [])   # Returns list[any] type
    
    print("Name: ", name, "\n")
    print("Age: ", age, "\n")
    print("City: ", city, "\n")
}

Run result:


Name: Alice
Age: 30
City: Unknown
Using with List Index

The .get() method can be used in combination with list index access to handle nested JSON data:


def main(args:list[str]) {
    # Simulate API streaming response parsing
    json_str:str = "{\"choices\": [{\"delta\": {\"content\": \"Hello\"}}]}"
    data_dict:dict = json_loads(json_str)
    
    # Get nested data, return type determined by default value
    if "choices" in data_dict and data_dict["choices"] {
        delta:dict[str, any] = data_dict["choices"][0].get("delta", {})
        content:str = delta.get("content", "")
        print("Content: ", content, "\n")
    }
}

Run result:


Content: Hello

9.6 TCP/UDP Binary Data Communication

TCP and UDP communication are essentially byte stream transmission. The write, sendto, and recvfrom functions all support bytes type data. The following are specific applications of bytes in network communication:

TCP Sending Binary Data


def main(args:list[str]) {
    client:TCPSocket = tcpsocket()
    client.connect("192.168.1.100", 8080)

    # Send custom binary protocol
    # Frame format: | Header(2) | Length(2) | Data(N) | Checksum(1) |
    header:bytes = b"\xAA\x55"
    length:bytes = b"\x00\x08"
    data:bytes = b"\x01\x02\x03\x04\x05\x06\x07\x08"
    checksum:bytes = b"\x2C"

    packet:bytes = header + length + data + checksum

    # Send binary data
    client.write(packet)
    print("Sent ")
    print(len(packet))
    print(" bytes\n")

    # Receive response
    response:str = client.read(4096)
    print("Response length: ")
    print(len(response))
    print("\n")

    client.close()
}

TCP Server Handling Binary Protocol


def main(args:list[str]) {
    server:TCPSocket = tcpsocket()
    server.bind("0.0.0.0", 8888)
    server.listen(128)
    print("TCP Server listening on port 8888\n")

    conn:TCPClient = server.accept()
    print("Client connected\n")

    # Receive client data
    data:str = conn.read(1024)
    print("Received ")
    print(len(data))
    print(" bytes\n")

    # Parse binary protocol
    # Frame format: | Magic(2) | Command(1) | Sequence(2) | Data Length(2) | Data(N) |
    if len(data) >= 7 {
        # Verify frame header (0xAA 0x55)
        print("Protocol header verified\n")

        # Send response
        response_header:bytes = b"\xAA\x55"
        response_cmd:bytes = b"\x81"  # Response command
        response_seq:bytes = b"\x00\x01"
        response_data:bytes = b"\x00\x00\x00\x00"

        response:bytes = response_header + response_cmd + response_seq + response_data
        conn.write(response)
        print("Response sent\n")
    }

    conn.close()
    server.close()
}

UDP Sending Binary Data


def main(args:list[str]) {
    sock:UDPSocket = udpsocket()
    sock.bind("127.0.0.1", 0)

    # Send custom binary protocol
    # Packet format: | Source Port(2) | Dest Port(2) | Length(2) | Checksum(2) | Data(N) |
    src_port:bytes = b"\x00\x00"
    dst_port:bytes = b"\x23\x28"  # Hexadecimal of 9000
    length:bytes = b"\x00\x0C"
    checksum:bytes = b"\x00\x00"
    payload:bytes = b"\x01\x02\x03\x04\x05\x06"

    packet:bytes = src_port + dst_port + length + checksum + payload

    # Send binary data
    sock.sendto(packet, "127.0.0.1", 9000)
    print("UDP packet sent\n")

    sock.close()
}

Network Byte Order Conversion

Network protocols typically use Big Endian byte order. Here is an example of handling multi-byte data:


def main(args:list[str]) {
    # Construct 32-bit length field
    b0:bytes = b"\x00"
    b1:bytes = b"\x00"
    b2:bytes = b"\x01"
    b3:bytes = b"\x00"

    length_32:bytes = b0 + b1 + b2 + b3

    print("32-bit length: ")
    print(len(length_32))
    print(" bytes\n")

    # Assemble complete frame
    header:bytes = b"\xAA\x55"
    frame:bytes = header + length_32

    print("Complete frame: ")
    print(len(frame))
    print(" bytes\n")
}

9.7 WebSocket

WebSocket is a protocol for full-duplex communication over a single TCP connection, suitable for application scenarios requiring high real-time performance, such as chat, games, real-time data push, etc. CatBase provides complete WebSocket client support.

websocket

websocket(url:str, headers:dict) : WebSocket - Create WebSocket connection

Creates a WebSocket client connection to the specified server.

Parameter description:

Parameter Type Required Default Description
url str Yes - WebSocket server address (e.g., ws://example.com/ws)
headers dict No None Optional request headers dictionary

Return value:

  • Returns a WebSocket type object for communicating with the server

Basic usage:


def main(args:list[str]) {
    # Create WebSocket connection
    ws:WebSocket = websocket("ws://127.0.0.1:8080/ws", None)

    # Send message
    ws.send("Hello, Server!")

    # Receive message
    msg:str = ws.recv()
    print("Received: ", msg, "\n")

    # Close connection
    ws.close()
}

WebSocket connection with authentication headers:


def main(args:list[str]) {
    # Set authentication headers
    headers:dict = {}
    headers["Authorization"] = "Bearer your-token-here"
    headers["X-App-Id"] = "my-app"

    ws:WebSocket = websocket("wss://example.com/ws", headers)

    ws.send("Hello with auth!")
    ws.close()
}

WebSocket Methods

Method Description
ws.send(message) Send message to server (multi-type support, see below)
ws.recv() : str Blocking receive of server message, returns empty string if connection is closed
ws.close() Close WebSocket connection

ws.send() Multi-type Support

ws.send() accepts multiple parameter types and automatically selects the WebSocket frame type based on the content:

Supported Types:

Parameter CatBase Type WebSocket Frame Opcode
"Hello" str text frame 0x81
str_var str text frame 0x81
b"Hello" bytes (printable ASCII) text frame (auto-convert) 0x81
b"Hello\nWorld" bytes (with \n) text frame (auto-convert) 0x81
b"\x00\xFF" bytes (with control/non-ASCII) binary frame 0x82
bytes_var bytes auto-detect smart
bytes_alloc(64) bytes auto-detect smart

Auto-detection Rules (for bytes type):

Byte Range Handling Frame
0x09 (\t), 0x0A (\n), 0x0D (\r) treated as printable text
0x20-0x7E printable ASCII text
0x7F (DEL) non-printable binary
0x80-0xFF (non-ASCII) non-printable binary
0x00-0x08, 0x0B, 0x0C, 0x0E-0x1F non-printable binary

Basic Usage:


def main(args:list[str]) {
    ws:WebSocket = websocket("ws://127.0.0.1:8080/ws", None)

    # Method 1: send str (text frame)
    ws.send("Hello, Server!")

    # Method 2: send bytes literal (auto-detect)
    ws.send(b"Hello")              # all ASCII → text frame
    ws.send(b"Line1\nLine2")       # with newline → text frame

    # Method 3: send binary data (binary frame)
    ws.send(b"\x00\xFF\x42")       # with control chars → binary frame
    ws.send(bytes_alloc(64))       # runtime alloc → auto-detect

    # receive
    msg:str = ws.recv()
    print("Received: ", msg, "\n")

    ws.close()
}

Comparison with Python websockets library:

Python websockets CatBase (auto-detect)
await ws.send("text") → text frame ws.send("text")0x81
await ws.send(b"text")binary frame ws.send(b"text")0x81 (auto text)
await ws.send(b"\x00\xFF") → binary frame ws.send(b"\x00\xFF")0x82
no smart detection CatBase is smarter

Notes:

  • str type is always sent as text frame
  • bytes type is auto-detected based on content
  • For explicit binary frame, send bytes containing non-ASCII chars to trigger it

WebSocket Client Example


# WebSocket audio client example
def audio_sender_task(ws:WebSocket) {
    print("[INFO] Send task started\n")

    # Open recording stream
    rec_stream:RecordStream = recordStream(
        rate="16000",
        channels="1",
        chunk="512",
        callback=on_audio_data
    )
    rec_stream.start_recording()

    # Send Hello message
    hello_msg:str = ""
    hello_msg = hello_msg + "{\"type\":\"hello\",\"version\":1,"
    hello_msg = hello_msg + "\"features\":{\"mcp\":true},"
    hello_msg = hello_msg + "\"transport\":\"websocket\"}"
    ws.send(hello_msg)

    # Loop to send audio data
    i:int = 0
    while i < 100 {
        # Simulate PCM data
        pcm_data:bytes = ""
        j:int = 0
        while j < 320 {
            pcm_data = pcm_data + "\x00\x00"
            j = j + 1
        }

        # Send audio frame
        ws.send(pcm_data)
        sleep(0.02)
        i = i + 1
    }

    rec_stream.stop_recording()
    rec_stream.close()
}

def on_audio_data(data:bytes) {
    print("Recording data: ", len(data), " bytes\n")
}

def main(args:list[str]) {
    print("========================================\n")
    print("      WebSocket Audio Client\n")
    print("========================================\n")

    # Connect to WebSocket server
    ws:WebSocket = websocket("ws://127.0.0.1:8080/ws", None)

    # Create send and receive threads
    thread audio_sender_task(ws)
    thread audio_receiver_task(ws)

    # Wait for a while
    sleep(10)

    print("[INFO] Closing connection...\n")
    ws.close()

    print("========================================\n")
    print("[INFO] WebSocket client closed\n")
    print("========================================\n")
}

Comparison of WebSocket with Python websocket-client

CatBase Python (websocket-client)
websocket(url, None) websocket.create_connection(url)
websocket(url, headers) websocket.create_connection(url, headers=headers)
ws.send(message) ws.send(message)
ws.recv() ws.recv()
ws.close() ws.close()

10. Multithreading Programming

Chapter Overview: Network programming lets us communicate with the outside world; this chapter introduces multithreading programming. Multithreading is an important means of improving program performance in modern programming. CatBase provides concise multithreading support, including thread creation, synchronization (Mutex), atomic operations, and other features.

10.1 Creating Threads

CatBase supports two ways to create threads:

Method 1: As a standalone statement (does not wait for the thread to finish)

thread func(args) - Creates a new thread and returns immediately without waiting for the thread to finish


def worker(id:int) {
    print("Worker ", id, " started\n")
    sleep(1)
    print("Worker ", id, " finished\n")
}

def main(args:list[str]) {
    print("Main thread started\n")
    
    thread worker(1)
    thread worker(2)
    thread worker(3)
    
    sleep(2)
    print("Main thread finished\n")
}

Method 2: As an expression (obtain a thread handle and can wait)

t:Thread = thread worker(args) - Creates a thread and returns a Thread handle


def worker(a:int, b:int) {
    print("Worker started with", a, b)
    result:int = a + b
    print("Worker result:", result)
}

def main(args:list[str]) {
    print("=== Testing thread() function ===\n")
    
    # Create a thread and get its handle
    t1:Thread = thread worker(1, 2)
    print("Thread created, t1 type:", type(t1), "\n")
    
    # Use join() to wait for the thread to finish
    t1.join()
    print("Thread t1 joined\n")
    
    # Create multiple threads
    t2:Thread = thread worker(10, 20)
    t2.join()
    print("Thread t2 joined\n")
    
    print("=== All thread tests passed ===\n")
}

Execution result:


=== Testing thread() function ===
Thread created, t1 type: Thread
Worker started with 1 2
Worker result: 3
Thread t1 joined
Worker started with 10 20
Worker result: 30
Thread t2 joined
=== All thread tests passed ===

Note:

  • When thread worker(args) is used as a statement, the thread starts immediately and runs in the background
  • When thread worker(args) is used as an expression, it returns a Thread handle that can be used with the .join() method to wait for the thread to finish
  • The return type of a function called by a thread must be void (no return value)

10.2 Thread Synchronization

mutex

mutex() : Mutex - Creates a mutex


def main(args:list[str]) {
    m:Mutex = mutex()
    print("Mutex created\n")
}

lock

m.lock() - Acquires the mutex (method form)


def worker(id:int, m:Mutex) {
    m.lock()
    print("Worker ", id, " locked\n")
    sleep(1)
    m.unlock()
}

unlock

m.unlock() - Releases the mutex (method form)


def worker(id:int, m:Mutex) {
    m.lock()
    print("Worker ", id, " working\n")
    m.unlock()
    print("Worker ", id, " unlocked\n")
}

Mutex Complete Example


def worker(id:int, m:Mutex) {
    m.lock()
    print("Worker ", id, " locked\n")
    sleep(1)
    print("Worker ", id, " unlocking\n")
    m.unlock()
}

def main(args:list[str]) {
    m:Mutex = mutex()
    
    # Use thread() as an expression to obtain thread handles
    t1:Thread = thread worker(1, m)
    t2:Thread = thread worker(2, m)
    t3:Thread = thread worker(3, m)
    
    # Wait for all threads to finish
    t1.join()
    t2.join()
    t3.join()
    
    print("All workers finished\n")
}

Purpose of Mutex

Mutex (mutual exclusion lock) is a synchronization mechanism used in multithreaded programming. In multithreaded programs, when multiple threads access shared resources (such as variables, files, database connections, etc.) at the same time, race conditions and data inconsistency problems may arise.

The role of a Mutex is to:

  1. Ensure that only one thread at a time can access shared resources
  2. Prevent data races: When one thread is modifying data, other threads must wait
  3. Protect critical sections: The region of code that accesses shared resources is called a critical section; a mutex is used to protect critical sections

Example scenario - The counter problem:

Suppose multiple threads perform addition on the same counter simultaneously:

Without a mutex:


Thread A reads counter=0
Thread B reads counter=0  (A has not written back yet)
Thread A writes back counter=1
Thread B writes back counter=1  (overwrites A's result, losing one addition)
Result: counter=1 (should be 2)

With a mutex:


Thread A acquires the mutex lock
Thread A reads counter=0
Thread A computes counter=1
Thread A writes back counter=1
Thread A releases the mutex lock

Thread B acquires the mutex lock
Thread B reads counter=1
Thread B computes counter=2
Thread B writes back counter=2
Thread B releases the mutex lock

Result: counter=2 (correct)
Feature Description
Purpose Protect shared resources and prevent data races
Characteristics Exclusive, non-reentrant (the same thread cannot lock it repeatedly)
Applicable scenarios Multiple threads accessing the same variable, file, database, or other shared resources

10.3 Thread-Safe Counter (Mutex Implementation)

CatBase does not provide atomic operation functions (such as atomic_add), but thread-safe counter operations can be implemented through a Mutex.

Using Mutex to Protect Shared Variables


# Global counter and mutex
counter: int = 0
m: Mutex = mutex()

def increment(id: int) {
    i: int = 0
    while i < 1000 {
        m.lock()
        counter = counter + 1
        m.unlock()
        i = i + 1
    }
    print("Worker ", id, " finished\n")
}

def main(args:list[str]) {
    # Start 3 threads incrementing the counter simultaneously
    t1: Thread = thread increment(1)
    t2: Thread = thread increment(2)
    t3: Thread = thread increment(3)

    t1.join()
    t2.join()
    t3.join()

    print("Final counter: ", counter, "\n")
}

Execution result:


Worker 1 finished
Worker 2 finished
Worker 3 finished
Final counter: 3000

Mutex vs. Atomic Operations

Feature Atomic Operations (not supported by CatBase) Mutex (supported by CatBase)
Performance Faster, lower overhead Slower, has locking overhead
Applicable scenarios Simple numeric operations Complex critical sections
Usage Counters, flags, etc. Protecting code blocks

Usage notes:

  1. When accessing shared variables in a multithreaded environment, you must use a Mutex for protection
  2. The scope of the lock should be as small as possible, only protecting the necessary critical section code
  3. Avoid deadlocks: do not attempt to acquire the same lock while already holding it

10.4 Waiting for Threads

join

t.join() - Waits for the thread to finish (method of the Thread object)


def worker(id:int) {
    print("Worker ", id, " started\n")
    sleep(1)
    print("Worker ", id, " finished\n")
}

def main(args:list[str]) {
    t1:Thread = thread worker(1)
    t2:Thread = thread worker(2)
    
    t1.join()
    t2.join()
    
    print("All workers finished\n")
}

Execution result:


Worker 1 started
Worker 2 started
Worker 1 finished
Worker 2 finished
All workers finished

Note:

  • t.join() is a method of the Thread object, used to wait for the thread to finish executing
  • If the thread has already finished, calling join() returns immediately

10.5 Message Queues

A message queue is an inter-thread communication mechanism used to safely pass data between threads. CatBase's message queue API is aligned with Python's queue.Queue.

queue

queue(maxsize:int) : Queue - Creates a message queue


def main(args:list[str]) {
    # Create a queue with a maximum capacity of 10
    q:Queue = queue(10)
    print("Queue created, maxsize: 10\n")
}

put_nowait

q.put_nowait(item:int) - Non-blocking put data into the queue (aligned with Python's queue.put_nowait())


def producer(q:Queue) {
    i:int = 0
    while i < 5 {
        msg:int = i * 100
        q.put_nowait(msg)
        print("Producer: put_nowait ", msg, "\n")
        i = i + 1
    }
}

get

q.get(timeout_ms:int) : int - Retrieves data from the queue with timeout support (aligned with Python's queue.get())

  • timeout_ms = 0: Non-blocking, returns 0 on timeout
  • timeout_ms = -1: Wait indefinitely
  • timeout_ms > 0: Timeout duration (in milliseconds)

def consumer(q:Queue) {
    i:int = 0
    while i < 5 {
        if !q.empty() {
            msg:int = q.get(100)  # Wait 100ms
            print("Consumer: get ", msg, "\n")
        } else {
            print("Consumer: queue is empty, waiting...\n")
        }
        i = i + 1
    }
}

empty

q.empty() : bool - Checks whether the queue is empty


if q.empty() {
    print("Queue is empty\n")
}

full

q.full() : bool - Checks whether the queue is full


if q.full() {
    print("Queue is full\n")
}

qsize

q.qsize() : int - Gets the number of elements in the queue (aligned with Python's queue.qsize())


size:int = q.qsize()
print("Queue size: ", size, "\n")

get_nowait

q.get_nowait() : int - Non-blocking retrieval from the queue; returns 0 when empty (aligned with Python's queue.get_nowait())


if !q.empty() {
    msg:int = q.get_nowait()
    print("Got: ", msg, "\n")
}

get_maxsize

q.get_maxsize() : int - Gets the maximum capacity of the queue (aligned with Python's queue.maxsize attribute)


max:int = q.get_maxsize()
print("Queue maxsize: ", max, "\n")

task_done

q.task_done() - Marks a task as done (aligned with Python's queue.task_done())


def consumer(q:Queue) {
    i:int = 0
    while i < 5 {
        if !q.empty() {
            msg:int = q.get_nowait()
            print("Consumer: got ", msg, "\n")
            q.task_done()  # Mark task as done
        }
        i = i + 1
    }
}

join

q.join() - Waits for all tasks to complete (aligned with Python's queue.join())


# Start producer and consumer
thread producer(q)
thread consumer(q)

# Wait for all tasks to complete
q.join()
print("All tasks completed!\n")

Compatible Legacy Functions

The following functions are legacy APIs that are still usable:

Function Description Equivalent method
queue_push(q, item) Push into queue q.put_nowait(item)
queue_pop(q) Pop data q.get(0)
queue_empty(q) Check empty q.empty()
queue_full(q) Check full q.full()
queue_size(q) Get size q.qsize()

Python queue API vs. CatBase API Comparison Table

Python CatBase Description
queue.Queue(maxsize) queue(maxsize) Create queue
q.put_nowait(item) q.put_nowait(item) Non-blocking put ✅
q.get(timeout) q.get(timeout_ms) Get with timeout ✅
q.get_nowait() q.get_nowait() Non-blocking get ✅
q.empty() q.empty() Check empty ✅
q.full() q.full() Check full ✅
q.qsize() q.qsize() Get size ✅
q.maxsize q.get_maxsize() Get max capacity ✅
q.task_done() q.task_done() Mark task as done ✅
q.join() q.join() Wait for completion ✅

Message Queue Complete Example


def producer(q:Queue) {
    i:int = 0
    while i < 5 {
        msg:int = i * 100
        q.put_nowait(msg)
        print("Producer: put_nowait ", msg, "\n")
        sleep(1)
        i = i + 1
    }
}

def consumer(q:Queue) {
    i:int = 0
    while i < 5 {
        if !q.empty() {
            msg:int = q.get(100)
            print("Consumer: get ", msg, "\n")
        } else {
            print("Consumer: queue is empty, waiting...\n")
        }
        sleep(1)
        i = i + 1
    }
}

def main(args:list[str]) {
    print("Testing Queue (Python-like API)...\n")

    # Create a queue with a maximum capacity of 10
    q:Queue = queue(10)

    print("Queue created, maxsize: 10\n")
    print("q.empty(): ", q.empty(), "\n")
    print("q.full(): ", q.full(), "\n")
    print("q.qsize(): ", q.qsize(), "\n")

    # Start producer and consumer
    thread producer(q)
    thread consumer(q)

    # Wait a while for the threads to complete
    sleep(6)

    print("\nFinal q.qsize(): ", q.qsize(), "\n")
    print("Test completed!\n")
}

Execution result:


Testing Queue (Python-like API)...

Queue created, maxsize: 10

q.empty():  true 

q.full():  false 

q.qsize():  0 

Producer: put_nowait  0 

Consumer: get  0 

Producer: put_nowait  100 

Consumer: get  100 

Producer: put_nowait  0 

Consumer: get  0 

Producer: put_nowait  100 

Consumer: get  100 

Producer: put_nowait  200 

Consumer: get  200 

Producer: put_nowait  300 

Consumer: get  300 

Producer: put_nowait  400 

Final q.qsize():  1 

Test completed!

Message Queue vs. Mutex Comparison

Feature Message Queue Mutex + Shared Variable
Data passing Passed automatically via the queue Passed manually via shared variables
Synchronization method Automatic (queue operations are atomic) Requires manual lock/unlock
Safety Higher, no deadlock Possible deadlock
Performance Slight overhead (queue operations) Better (only lock operations)
Applicable scenarios Producer-consumer pattern Critical section protection

Uses of message queues:

  1. Producer-consumer pattern: One thread produces data, another thread consumes data
  2. Task dispatching: The main thread dispatches tasks to worker threads
  3. Event notification: Passing events or messages between threads
  4. Decoupling: Allowing producers and consumers to be unaware of each other's existence

10.6 Graceful Shutdown Events

The graceful shutdown event (ShutdownEvent) is an inter-thread communication mechanism, similar to Python's asyncio.Event, used to implement graceful program shutdown.

Basic Concepts

When the program receives a shutdown signal (such as Ctrl+C), we need a mechanism to:

  1. Notify all worker threads of the shutdown request
  2. Worker threads check the shutdown status and stop in an orderly manner
  3. The main thread waits for all worker threads to complete

init_shutdown_event

init_shutdown_event() - Initializes the shutdown event (automatically called before main starts)


# No need to call manually; it is automatically initialized when the program starts

set_shutdown_event

set_shutdown_event() - Sets the shutdown event (equivalent to Python's event.set())


# Set the shutdown event, notifying all threads that the program is about to shut down
set_shutdown_event()

is_shutdown_event_set

is_shutdown_event_set() - Checks whether the shutdown event is set; returns a boolean value


if is_shutdown_event_set() {
    print("Shutdown requested\n")
}

shutdown_requested

shutdown_requested() - Checks whether shutdown is requested; returns a boolean value (equivalent to is_shutdown_event_set())


# Periodically check in the worker thread
for {
    if shutdown_requested() {
        print("Stopping task...\n")
        return
    }
    # Continue working
}

wait_shutdown_event

wait_shutdown_event(timeout_seconds:float) - Waits for the shutdown event or timeout

  • No parameter or parameter is 0: wait indefinitely
  • Parameter > 0: timeout after the specified number of seconds

# Wait for the shutdown event (wait up to 10 seconds)
result:bool = wait_shutdown_event(10)
if result {
    print("Shutdown event received!\n")
} else {
    print("Timeout reached\n")
}

Complete Example


async def long_running_task(id:int) {
    print("Task ", id, " started\n")
    for {
        # Periodically check shutdown requests
        if shutdown_requested() {
            print("Task ", id, " stopping...\n")
            return
        }
        sleep(0.1)  # Simulate work
    }
}

def main(args:list[str]) {
    print("Starting tasks...\n")

    # Start multiple worker threads
    for i in range(5) {
        thread long_running_task(i)
    }

    # Wait for the shutdown event (wait up to 30 seconds)
    print("Waiting for shutdown (Ctrl+C to trigger)...\n")
    result:bool = wait_shutdown_event(30)

    if result {
        print("Shutdown event received!\n")
    } else {
        print("Timeout, continuing...\n")
    }

    print("Main program finished\n")
}

Execution result (press Ctrl+C):


Starting tasks...
Task 0 started
Task 1 started
Task 2 started
...
Waiting for shutdown (Ctrl+C to trigger)...
^C
Task 0 stopping...
Task 1 stopping...
Task 2 stopping...
Shutdown event received!
Main program finished

Comparison with Python asyncio.Event:

CatBase Python asyncio
init_shutdown_event() asyncio.Event()
set_shutdown_event() event.set()
is_shutdown_event_set() event.is_set()
shutdown_requested() event.is_set()
wait_shutdown_event(timeout) await event.wait(timeout)

10.7 Thread-Safe Queue (Queue)

In multithreaded programming, threads often need to pass data to each other. CatBase provides a thread-safe queue (Queue), aligned with Python's queue.Queue API.

queue

queue(maxsize:int) : Queue - Creates a thread-safe queue

Creates a thread-safe queue for passing data between threads.

Parameters:

Parameter Type Required Default value Description
maxsize int No 0 Maximum queue capacity; 0 means unlimited

Return value:

  • Returns a Queue type object

Basic usage:


def producer(q:Queue) {
    i:int = 0
    while i < 5 {
        q.put_nowait(i)
        print("Produced: ", i, "\n")
        i = i + 1
    }
}

def consumer(q:Queue) {
    i:int = 0
    while i < 5 {
        value:int = q.get(-1)  # Wait indefinitely
        print("Consumed: ", value, "\n")
        i = i + 1
    }
}

def main(args:list[str]) {
    q:Queue = queue(0)  # Unlimited queue

    thread producer(q)
    thread consumer(q)

    sleep(2)
}

Queue Methods

Method Description
q.put_nowait(item:int) Non-blocking put of item; silently dropped when full
q.get(timeout_ms:int) : int Get with timeout; returns 0 on timeout; timeout_ms=-1 means wait indefinitely
q.get_nowait() : int Non-blocking get; returns 0 when empty
q.empty() : bool Check whether the queue is empty
q.full() : bool Check whether the queue is full
q.qsize() : int Get the number of elements in the queue
q.maxsize() : int Get the maximum capacity of the queue
q.task_done() Mark a task as done
q.join() Wait for all tasks to complete

Parameter details:

  • timeout_ms values:
    • 0: Non-blocking; returns 0 immediately if the queue is empty
    • -1: Wait indefinitely until the queue has data
    • > 0: Wait for the specified number of milliseconds; returns 0 on timeout

Queue Usage Example


def worker(id:int, q:Queue) {
    # Wait indefinitely to get data
    value:int = q.get(-1)
    print("Worker ", id, " got: ", value, "\n")

    # Mark the task as done
    q.task_done()
}

def main(args:list[str]) {
    q:Queue = queue(10)  # Up to 10 elements

    # Start consumer threads
    thread worker(1, q)
    thread worker(2, q)

    # Producer puts data
    i:int = 0
    while i < 5 {
        q.put_nowait(i)
        print("Produced: ", i, "\n")
        sleep(0.1)
        i = i + 1
    }

    # Wait for all tasks to complete
    q.join()

    print("All tasks completed\n")
}

Queue vs. Python queue.Queue Comparison

CatBase Python
queue(maxsize) queue.Queue(maxsize)
q.put_nowait(item) q.put_nowait(item)
q.get(timeout_ms) q.get(timeout=timeout_ms/1000)
q.get_nowait() q.get_nowait()
q.empty() q.empty()
q.full() q.full()
q.qsize() q.qsize()
q.task_done() q.task_done()
q.join() q.join()

11. Signal Handling

Chapter Overview: This chapter introduces the signal handling mechanism. Signals are brief asynchronous notifications sent by the operating system to a process to report events (such as Ctrl+C interrupts). CatBase implements graceful shutdown through the ShutdownEvent mechanism, aligned with Python's signal handling approach.

11.1 Basic Concepts

Signal is a simple form of inter-process communication, sent by the operating system kernel to a process. Common signals include:

Signal Value Description
SIGINT 2 Interrupt signal (Ctrl+C)
SIGTERM 15 Termination signal (the default kill signal)
SIGKILL 9 Forced termination signal (cannot be caught)
SIGHUP 1 Hangup signal (terminal closed)

11.2 CatBase's Signal Handling Mechanism

CatBase does not directly expose the underlying signal API (such as signal(), sigaction()); instead, it handles common graceful shutdown scenarios through the high-level ShutdownEvent mechanism.

How It Works


┌─────────────────────────────────────────────────────────────┐
│                     CatBase Program                          │
│                                                             │
│  ┌──────────────┐     SIGINT/SIGTERM      ┌──────────────┐ │
│  │  Operating    │ ───────────────────────►│   Signal      │ │
│  │  System       │       Ctrl+C            │   Handling    │ │
│  │  Kernel       │                         │   Hook        │ │
│  └──────────────┘                          └──────┬───────┘ │
│                                                    │         │
│                                                    ▼         │
│                                          ┌─────────────────┐ │
│                                          │ ShutdownEvent   │ │
│                                          │ .is_set = true  │ │
│                                          └────────┬────────┘ │
│                                                   │          │
│         ┌────────────────┬────────────────┬──────┘          │
│         ▼                ▼                ▼                │
│    ┌─────────┐     ┌─────────┐     ┌─────────┐              │
│    │ Thread1 │     │ Thread2 │     │ Thread3 │              │
│    │ Check   │     │ Check   │     │ Check   │              │
│    │ flag    │     │ flag    │     │ flag    │              │
│    └─────────┘     └─────────┘     └─────────┘              │
└─────────────────────────────────────────────────────────────┘

11.3 Comparison with Python

Python Signal Handling Example


import signal
import sys

def handle_signal(signum, frame):
    print("Received signal, shutting down...")
    sys.exit(0)

# Register signal handler
signal.signal(signal.SIGINT, handle_signal)
signal.signal(signal.SIGTERM, handle_signal)

# Main loop
while True:
    # Process tasks
    pass

CatBase Graceful Shutdown Example


async def worker(id:int) {
    print("Worker ", id, " started\n")
    while true {
        # Periodically check shutdown requests
        if shutdown_requested() {
            print("Worker ", id, " stopping...\n")
            return
        }
        sleep(0.1)
    }
}

def main(args:list[str]) {
    print("Starting workers...\n")

    # Start multiple worker threads
    for i in range(5) {
        thread worker(i)
    }

    # Wait for the shutdown event
    # When Ctrl+C or the kill command is received, the ShutdownEvent will be set
    wait_shutdown_event(0)  # 0 means wait indefinitely

    print("All workers stopped gracefully\n")
}

11.4 Relationship Between Signal Handling and ShutdownEvent

Aspect Python CatBase
Underlying mechanism signal.signal() Zig thread synchronization primitives
Coordination mechanism Callback functions ShutdownEvent flag
Worker thread check Set flag in callback shutdown_requested()
Wait mechanism time.sleep() loop wait_shutdown_event()

11.5 Practical Use Cases

Scenario 1: Graceful Server Shutdown


async def handle_request(client_id:int) {
    print("Handling request from client ", client_id, "\n")
    # Handle the request...
}

def main(args:list[str]) {
    print("Web server starting on port 8080...\n")

    # Start server threads
    for i in range(100) {  # Simulate 100 connections
        thread handle_request(i)
    }

    # Wait for the shutdown signal
    print("Press Ctrl+C to stop the server...\n")
    wait_shutdown_event(0)

    print("Shutting down server...\n")
    # Clean up resources...
}

Scenario 2: Scheduled Task Graceful Exit


async def scheduled_task(id:int) {
    print("Task ", id, " running\n")
    for {
        if shutdown_requested() {
            print("Task ", id, " cancelled\n")
            return
        }
        # Execute task...
        sleep(1)
    }
}

def main(args:list[str]) {
    # Start scheduled tasks
    for i in range(3) {
        thread scheduled_task(i)
    }

    # Automatically exit after running for 60 seconds
    result:bool = wait_shutdown_event(60)
    if result {
        print("Shutdown was triggered\n")
    } else {
        print("Timeout reached\n")
    }
}

11.6 Notes

  1. SIGKILL cannot be caught: When a process receives the SIGKILL signal, the operating system immediately terminates the process; no cleanup code can be executed
  1. Main thread waiting: It is recommended to use wait_shutdown_event() in the main thread to wait for the shutdown signal, giving worker threads time to check shutdown_requested() and exit in an orderly manner
  1. Check frequency: Worker threads should periodically check shutdown_requested(); it is recommended to check on every loop iteration or at fixed time intervals
  1. Blocking operations: If worker threads have blocking operations (such as network waits), a timeout should be set so that the thread can respond to shutdown requests in a timely manner

12. Configuration File Parsing

Chapter Overview: This chapter introduces how to use the Config type to read and parse INI-format configuration files, similar to Python's configparser. This is very useful for managing application configurations.

12.1 The Config Type

Config is a built-in object type in CatBase, used to parse INI-format configuration files.


config:Config = config("config.conf")

INI configuration file format:


# Comment
[section_name]
key1 = value1
key2 = value2

[another_section]
key = value

12.2 Loading a Configuration File

Use config(filename) to load a configuration file:


config:Config = config("config.conf")

If the file does not exist or cannot be parsed, an empty Config object is returned.

12.3 Getting Configuration Values

CatBase provides several functions to get configuration values:

Function Description Return type
config_get(config, section, key) Get a string value str
config_get_int(config, section, key, default) Get an integer value int
config_get_float(config, section, key, default) Get a float value float
config_has_key(config, section, key) Check whether a key exists bool

Example:


def main(args:list[str]) {
    # Load the configuration file
    config:Config = config("app.conf")

    # Get a string value
    websocket_url:str = config_get(config, "websocket", "url")
    print("URL: ", websocket_url, "\n")

    # Get an integer value (with a default value provided)
    port:int = config_get_int(config, "websocket", "port", 8080)
    print("Port: ", port, "\n")

    # Check whether a key exists
    if config_has_key(config, "websocket", "ssl") {
        print("SSL is configured\n")
    }
}

12.4 Complete Example

Suppose the app.conf file contains the following:


[websocket]
url = ws://localhost:8080
device_id = device001
port = 8080

[recording]
sample_rate = 16000
channels = 1

The corresponding CatBase code:


def main(args:list[str]) {
    print("=== Config Demo ===\n")

    config:Config = config("app.conf")

    # Read websocket configuration
    url:str = config_get(config, "websocket", "url")
    device_id:str = config_get(config, "websocket", "device_id")
    port:int = config_get_int(config, "websocket", "port", 8080)

    print("WebSocket URL: ", url, "\n")
    print("Device ID: ", device_id, "\n")
    print("Port: ", port, "\n")

    # Read recording configuration
    sample_rate:int = config_get_int(config, "recording", "sample_rate", 16000)
    channels:int = config_get_int(config, "recording", "channels", 1)

    print("Sample rate: ", sample_rate, "\n")
    print("Channels: ", channels, "\n")
}

Execution result:


=== Config Demo ===

WebSocket URL:  ws://localhost:8080
Device ID:  device001
Port:  8080
Sample rate:  16000
Channels:  1

12.5 Comparison with Python configparser

CatBase Python
config("file.conf") configparser.ConfigParser() + read()
config_get(c, "sec", "key") c.get("sec", "key")
config_get_int(c, "sec", "key", 0) c.getint("sec", "key", fallback=0)
config_get_float(c, "sec", "key", 0.0) c.getfloat("sec", "key", fallback=0.0)
config_has_key(c, "sec", "key") c.has_option("sec", "key")

13. Coroutine Programming

Chapter Overview: This chapter introduces coroutine programming. A coroutine is a lightweight concurrency model, lighter than a thread, suitable for high-concurrency server application scenarios. CatBase provides async def and await syntax to support coroutine programming.

13.1 Defining Coroutine Functions

async def

Use the async def keyword to define a coroutine function:


async def fetch_data(url:str) -> str {
    print("Fetching data from ", url, "\n")
    response:str = "Data from " + url
    return response
}

async def process_request(req_id:int) {
    result:str = await fetch_data("http://example.com/api/" + str(req_id))
    print("Request ", req_id, " completed: ", result, "\n")
}

Description:

  • async def defines a coroutine function
  • Inside a coroutine function, the await keyword can be used to wait for other coroutines to complete

13.2 Awaiting a Coroutine

await

Use the await keyword to wait for the result of a coroutine's execution:


async def get_data() -> str {
    await sleep(1)
    return "Data loaded"
}

def main(args:list[str]) {
    result:str = await get_data()
    print("Result: ", result, "\n")
}

Description:

  • await can only be used inside an async def function
  • await pauses the current coroutine, waiting for the target coroutine to complete and return its result

13.3 Relationship Between Coroutines and Threads

CatBase's async/await syntax:

  • Functions defined with async def execute in a separate thread
  • An await call creates a new thread and joins it to wait for completion
  • async/await preserves concise syntax similar to Python's, but achieves true multithreading parallelism

Core Feature Comparison

Feature thread async (coroutine)
Creation overhead Higher (requires OS to allocate resources) Lower (uses a thread pool underneath)
Memory usage About 1MB per thread About 1KB per coroutine (actually uses threads)
Switching cost OS context switch Thread join synchronization
Quantity limit Limited by memory, typically thousands Limited by the number of threads
Synchronization method Mutex lock, atomic operations await synchronous wait
True parallelism Yes Yes (underlying is thread)

Application Scenario Comparison

1. I/O-intensive vs. CPU-intensive

Scenario Recommended approach Reason
Network requests, file read/write, database queries Coroutine (async) I/O wait does not occupy CPU; many coroutines can run concurrently
Video encoding, image processing, scientific computing Thread (thread) Requires continuous CPU computation; can leverage multiple cores

2. Concurrency quantity

Scenario Recommended approach Reason
Need to handle hundreds to thousands of tasks simultaneously Either works async/await uses threads underneath, with good performance
Need to handle a large number of I/O-intensive tasks simultaneously Coroutine (async) Concise code, clear syntax
Need CPU-intensive parallel computing Thread (thread) Direct thread control, more flexible

3. Programming complexity

Scenario Recommended approach Reason
Simple concurrent tasks Coroutine (async) Concise code, no need to worry about lock issues
Need shared state Thread (thread) Mature synchronization mechanism
Complex dependency relationships Coroutine (async) await syntax is more intuitive

13.4 Coroutine Usage Examples

Example 1: High-Concurrency Web Server


async def handle_client(client_id:int) {
    # Simulate handling a client request
    print("Client ", client_id, " connected\n")
    
    # Simulate network latency
    await sleep(1)
    
    print("Client ", client_id, " request processed\n")
}

def main(args:list[str]) {
    # Simulate the server handling 10000 client connections simultaneously
    for i in range(10000) {
        handle_client(i)
    }
    
    print("Server started, handling 10000 clients\n")
}

Execution result:


Server started, handling 10000 clients
Client 0 connected
Client 1 connected
Client 2 connected
...
Client 9999 connected
Client 0 request processed
Client 1 request processed
...

Example 2: Concurrent Network Requests


async def fetch_url(url:str) -> str {
    # Simulate a network request
    await sleep(1)
    return "Response from " + url
}

async def crawl_urls() {
    urls:list[str] = ["http://a.com", "http://b.com", "http://c.com"]
    
    results:list[str] = []
    for url in urls {
        result:str = await fetch_url(url)
        results.append(result)
    }
    
    print("All fetched: ", results, "\n")
}

def main(args:list[str]) {
    crawl_urls()
}

Example 3: Concurrent Database Queries


async def query_user(user_id:int) -> str {
    # Simulate database query latency
    await sleep(0.5)
    return "User-" + str(user_id)
}

async def get_all_users() {
    # Concurrently query 100 users
    results:list[str] = []
    for i in range(100) {
        result:str = await query_user(i)
        results.append(result)
    }
    
    print("Total users: ", len(results), "\n")
}

def main(args:list[str]) {
    get_all_users()
}

13.5 Thread Usage Examples

Example 1: CPU-Intensive Computation


def calculate(start:int, end:int) -> int {
    sum:int = 0
    for i in range(start, end) {
        sum = sum + i * i
    }
    return sum
}

def main(args:list[str]) {
    # Use 4 threads for parallel computation
    thread calculate(0, 250000)
    thread calculate(250000, 500000)
    thread calculate(500000, 750000)
    thread calculate(750000, 1000000)
    
    sleep(1)
    print("Calculation completed\n")
}

Example 2: Background Task Processing


def background_task(task_id:int) {
    print("Task ", task_id, " started\n")
    sleep(2)
    print("Task ", task_id, " completed\n")
}

def main(args:list[str]) {
    # Start multiple background tasks
    for i in range(10) {
        thread background_task(i)
    }
    
    print("All tasks dispatched\n")
    sleep(3)
}

13.6 Selection Guide

Choose the appropriate concurrency model based on the following factors:

Scenarios for choosing coroutines (async/await):

  • High-concurrency network services (web servers, API services)
  • Need to handle a large number of I/O operations simultaneously
  • Need to handle tens of thousands to hundreds of thousands of concurrent connections
  • Pursue higher resource utilization efficiency

Scenarios for choosing threads (thread):

  • CPU-intensive tasks (computation, encryption, compression)
  • Need to leverage multi-core CPUs
  • Integration with existing threaded code
  • Fewer tasks but computationally heavy

Mixed usage:

  • The main server uses coroutines to handle high-concurrency connections
  • CPU-intensive tasks are handed off to a thread pool

14. Import System

Chapter Overview: Multithreading programming lets us fully utilize system resources; this chapter introduces the import system. CatBase supports importing external C language libraries (.so/.a files), allowing you to use the rich C language ecosystem. At the same time, CatBase also supports compiling code into shared libraries for use by other programs.

14.0 Import Path Resolution Rules

CatBase supports importing the following file types:

  1. Importing .cat files: import other CatBase source files and call the functions defined in them
  2. Importing .catc files: import package files (functionally identical to .cat, but indicates "this is a package to be imported by others")
  3. Importing .so files: import shared libraries and call C functions
  4. Importing .a files: import static libraries and call C functions

Statement Form

An import statement is single-line: a line break ends the statement. The whitespace separator after the import keyword must be followed by a valid path; every import may set an alias via as alias:


import <path> as <alias>      # with alias
import <path>                 # when alias is omitted, the compiler infers one from the path

import supports the following three unambiguous path forms:

  • Quoted string: import "audio.sub", import "./helper.cat", import "/usr/lib/libm.so"
  • Path starting with ./ or ../: import ./foo.cat as bar, import ../shared/utils as utils
  • Unquoted path form (IDENT/DIV/DOT concatenation): import audio.sub, import apkg.sub_pkg1.sub_pkg_1, import /abs/path/libfoo.so. Note: absolute paths may be unquoted; the lexer treats them as paths rather than division operators.

When quotes are required: when the path contains hyphens, spaces, or other special characters, quotes are mandatory: import "/usr/lib/x86_64-linux-gnu/libopus.so" as _opus_native.

One import statement may reference only 1 target file; for example, import a b is illegal.

Charset Constraint (Mixing Check)

The path must not mix . with /. The compiler validates as follows:

  • Absolute / relative path (contains /): apart from file extensions (such as .so, .cat, .catc), the path must not contain .; the relative-path markers .. and . are exempt
  • Package path (contains only ., no /): only allows . inside package-name segments (i.e. the audio.sub form); the entire path must not contain /

Violations cause a compile error, for example:


import "audio/sub.cat"     # error: package names cannot use slash style
import "audio.sub/foo"     # error: a single import must not mix . and /
import "audio./sub"        # error: a single import must not mix . and /

Three Path Categories

The path after import is classified as one of the following three forms, each following a different lookup logic:

Path form Example Resolution base
Absolute path (starts with /) /usr/lib/libm.so, /abs/path/libfoo.so Use the path as-is
Relative path (contains / but does not start with /) ./helper.cat, ../shared/utils, foo/bar Relative to the directory of the current .cat file
Package path / bare name (no /) audio.sub, opus, apkg.sub_pkg1.sub_pkg_1 Relative to the packages/ directory of the compiler's install root

1. Absolute path

Paths that start with / are loaded directly without any lookup:


import /usr/lib/x86_64-linux-gnu/libm.so as math
import "/opt/myapp/libcustom.so" as custom

2. Relative path

Paths that do not start with / but contain / (such as ./foo, ../foo, foo/bar) are joined relative to the directory of the current .cat file. A bare filename (e.g. tools.so) is also treated as a relative path resolving to a sibling of the current directory:


import ./helper.cat as helper              # helper.cat in the source file's directory
import ../shared/utils as utils            # shared/utils in the parent directory
import ./libmylib.so as mylib              # libmylib.so in the source file's directory
import tools.so as tools                   # tools.so in the current directory

3. Package path (.-separated)

Paths that contain only . and no / are looked up as "package names". The package name is split by . (e.g. audio.sub), mapping to nested subdirectories under packages/:


import audio                # → packages/audio/index.catc
import audio.sub            # → packages/audio/sub/index.catc
import apkg.sub_pkg1.sub_pkg_1   # → packages/apkg/sub_pkg1/sub_pkg_1/index.catc

Three-phase package lookup (in order):

  1. index.catc first: replace . with / and append /index.catc to the last segment as the package entry file
    • for example, audio.subpackages/audio/sub/index.catc
  2. libconfig.conf fallback: if no index.catc exists in that directory, look up libconfig.conf in the package root and locate an entry by its <alias>|<target>|<distro> three-part section name (see below)
  3. Last-segment .so direct load: if the package path's last segment is literally .so (including versioned forms like libc.so.6), skip the first two phases and load the file directly; . in the path is treated as a directory separator
    • for example, audio.test.sopackages/audio/test.so
    • for example, libc.so.6packages/libc.so.6

4. Bare name (top-level package)

A path with neither / nor . (such as opus or json_helper) is equivalent to import <name> and resolves only to packages/<name>/index.catc:


import opus as opus_lib        # → packages/opus/index.catc
import json_helper as jh       # → packages/json_helper/index.catc

If the package directory exists but has no index.catc, the compiler reports an error—for example, import "audio" fails when packages/audio/index.catc is missing, and the message prompts the user to "create index.catc in that directory".

Package Root Location

The package root is the packages/ directory at the parent level of where catcc lives; the real path is obtained by resolving symlinks through EvalSymlinks. bin/ and packages/ must be siblings, otherwise a compile error is reported:


<install-root>/
├── bin/
│   └── catcc
└── packages/
    ├── opus/
    │   └── index.catc
    ├── audio/
    │   ├── index.catc
    │   └── libconfig.conf
    └── ...

The libconfig.conf File

When a package path (containing . or a bare name) cannot find packages/<path>/index.catc, the compiler walks upward through the package directories looking for a libconfig.conf file and locates the entry by its <alias>|<target>|<distro> three-part section name:

File format:


[libc|x86_64-linux-gnu|ubuntu]
path=/usr/lib/x86_64-linux-gnu/libc.so.6
soname=libc.so.6

[libc|arm-linux-gnueabihf|raspbian]
path=/lib/arm-linux-gnueabihf/libc.so.6
soname=libc.so.6

[opus|x86_64-linux-gnu|ubuntu]
path=/usr/lib/x86_64-linux-gnu/libopus.so.0
soname=libopus.so.0

Field description:

  • Section name: <alias>|<target>|<distro>, three parts separated by |, respectively the import alias, target triple (e.g. x86_64-linux-gnu), and distro (e.g. ubuntu, debian, raspbian)
  • path: the actual library file path used at compile time
  • soname: the SONAME used at link / run time (passed to dlopen)

If no section matches, a compile error is reported including the package name, alias, target, distro, and full lookup path to aid debugging.

Suffix Rules in Paths

The v9 rules remove auto-completion of the .cat / .catc suffix. When referencing .cat, .catc, .so, or .a files, the suffix must be specified explicitly:

  • Referencing .cat / .catc package files: the suffix is mandatory (explicit suffixes are recommended to avoid ambiguity)
  • Referencing .so / .a library files: the suffix is mandatory

The wildcard * has been completely removed from import statements to avoid path-resolution ambiguity; you must write the full path or package name.

Usage Examples


# === Relative paths (source file's directory as base) ===
import ./helper.cat as helper
import ./util.catc as util
import ../shared/utils as utils
import ./libmylib.so as mylib
import ./libtest.a as test

# === Absolute paths ===
import /usr/lib/x86_64-linux-gnu/libm.so as math
import "/opt/myapp/libcustom.so" as custom

# === Package paths (contain .) ===
import audio as audio_lib                  # → packages/audio/index.catc
import audio.sub as audio_sub_lib          # → packages/audio/sub/index.catc
import apkg.sub_pkg1.sub_pkg_1 as lh       # → packages/apkg/sub_pkg1/sub_pkg_1/index.catc
import audio.test.so as at                 # → packages/audio/test.so (last segment .so direct load)

# === Bare names (top-level package) ===
import opus as opus_lib                    # → packages/opus/index.catc
import json_helper as jh                   # → packages/json_helper/index.catc

# === Quoted / unquoted equivalents ===
import "opus" as opus
import opus as opus
import opus                                # default alias = "opus"

Error handling: when a package path import "X.Y.Z" resolves to packages/<X>/<Y>/<Z>/ but that directory has neither index.catc nor a Z alias in any ancestor's libconfig.conf, the compiler reports an error that includes the package name, alias, target, distro, and full lookup path.

Compiler Parameters target and distro

When the v9 import resolver falls back to libconfig.conf alias lookup, it must use two compiler-level context parameters — target (target triple) and distro (target Linux distribution) — to correctly locate the .so file on the host system. Both values can be read from conf/config.conf as defaults and overridden by CLI flags.

Source and override

Parameter Config file field CLI override Default
target [compiler] target -target <arch-os-abi> x86_64-linux-gnu (auto-detected via uname -m when empty)
distro [compiler] distro -distro <name> ubuntu

Common target values (passed to Zig's -target):

  • x86_64-linux-gnu — mainstream Linux server/desktop (default)
  • aarch64-linux-gnu — ARMv8 Linux (Raspberry Pi 4 / ARM server / Kunpeng)
  • arm-linux-gnueabihf — ARMv7 32-bit Linux (Raspberry Pi 3 / embedded / old phones)
  • x86_64-windows-gnu — Windows MinGW (x86_64)
  • aarch64-macos-none — Apple Silicon Mac (M1 / M2 / M3)
  • x86_64-macos-none — Intel Mac
  • riscv64-linux-gnu — RISC-V 64-bit

distro value: since v9 the whitelist is removed — any string is accepted. The distro parameter is only used as the third match key when looking up a <alias>|<target>|<distro> section in libconfig.conf. As long as the package's conf/libconfig.conf has a matching section, anything you pass is valid.

For convenience, CatBase's bundled packages ship with the following distro identifiers as examples:

  • ubuntu — Ubuntu (Debian family, e.g. 20.04 / 22.04 / 24.04)
  • debian — Debian (stable / testing / unstable)
  • raspbian — Raspberry Pi OS (Debian-based, 32/64-bit Pi)
  • centos — CentOS (RHEL-compatible, 7 / 8 / 9 stream)
  • rhel — Red Hat Enterprise Linux (8 / 9)
  • arch — Arch Linux (rolling release)
  • suse — openSUSE (Leap / Tumbleweed)

Custom SDK / board: add a [<alias>|<target>|<my_distro>] section to your packages/<pkg>/conf/libconfig.conf, then pass -distro my_distro at compile time. For example, to support your own custom embedded board named myboard, add [libopus|arm-linux-gnueabihf|myboard] and run with -distro myboard. The my_distro value can be any string (pinyin, version number, chip codename, etc.) — it is only used as the section name match key.

Why target + distro Together Matter for Package Lookup

Every section in libconfig.conf uses the <alias>|<target>|<distro> three-part name. Only an entry whose target and distro both match is selected:


[opus|x86_64-linux-gnu|ubuntu]
path=/usr/lib/x86_64-linux-gnu/libopus.so.0
soname=libopus.so.0

[opus|arm-linux-gnueabihf|raspbian]
path=/usr/lib/arm-linux-gnueabihf/libopus.so.0
soname=libopus.so.0

Different distros install the same library to different paths and use different SONAMEs; different architectures use entirely different filenames. target answers "which CPU/OS instruction set" (it decides the CPU instructions, ABI, and dynamic-linker path prefix), while distro answers "which distro's package layout" (it decides whether the library is installed under /usr/lib or /lib, whether the SONAME is .so.6 or .so.6.0, and whether a versioned suffix is used). Both must match to uniquely identify the actual path and SONAME of a given .so on the host.

For example, import opus as opus_lib resolves to completely different .so files under different target/distro combinations:

Compiler flags Resolved file
-target x86_64-linux-gnu -distro ubuntu /usr/lib/x86_64-linux-gnu/libopus.so.0
-target aarch64-linux-gnu -distro ubuntu /usr/lib/aarch64-linux-gnu/libopus.so.0
-target arm-linux-gnueabihf -distro raspbian /usr/lib/arm-linux-gnueabihf/libopus.so.0
-target x86_64-linux-gnu -distro arch /usr/lib/libopus.so (Arch has no multi-arch subdir)

Error messages report the current target= and distro= together to make lookup failures easy to debug. For example:


package 'opus' alias 'opus' (target=x86_64-linux-gnu, distro=ubuntu) not found in
  /opt/catbase/packages/opus/libconfig.conf (imported in main.cat as 'opus_lib')

This means no section in libconfig.conf matches the current target/distro pair; either add a corresponding section or switch the -target / -distro flags.

CLI Usage

# Use the default target / distro from conf/config.conf
catcc main.cat

# Cross-compile to ARMv7 Raspberry Pi, matching the raspbian section
catcc -target arm-linux-gnueabihf -distro raspbian main.cat

# Override only distro; target remains the default x86_64-linux-gnu
catcc -distro debian main.cat

# Override only target; distro remains the default ubuntu
catcc -target aarch64-linux-gnu main.cat

Cross-platform best practice: in packages/<pkg>/libconfig.conf, pre-declare multiple [<alias>|<target>|<distro>] sections so the same CatBase source can import the same package on every target without any code change. Developers only need to switch -target / -distro to cross-compile seamlessly across Ubuntu, Debian, Raspberry Pi, ARM servers, and so on.

14.1 Generating Shared Libraries

CatBase supports compiling code into shared libraries (.so files) for import and use by other programs or CatBase code.

Compiling a Shared Library

Use the -shared parameter to compile CatBase code into a shared library:


# Compile to generate the shared library libmylib.so
catcc -shared mylib.cat

The generated shared library file name is libmylib.so (the lib prefix is automatically added).

Shared Library Example

First create a CatBase source file containing public functions:


# mylib.cat - Define public functions
def add(a:int, b:int) {
    result:int = a + b
    print("add: ", a, " + ", b, " = ", result, "\n")
}

def multiply(a:int, b:int) {
    result:int = a * b
    print("multiply: ", a, " * ", b, " = ", result, "\n")
}

Compile into a shared library:


catcc -shared mylib.cat

This will generate the libmylib.so file.

14.2 Importing Shared Libraries

After compiling and generating a shared library, you can import and use it in other CatBase code:


# main.cat - Use the shared library
import "./libmylib.so" as mylib

def main(args:list[str]) {
    mylib.add(5, 3)
    mylib.multiply(4, 7)
}

Compile and run:


catcc main.cat && ./main

Execution result:


add: 5 + 3 = 8
multiply: 4 * 7 = 28

14.3 Importing C Libraries

CatBase supports importing shared libraries (.so) and static libraries (.a) written in C.

Importing a .so File


import "./libtest.so" as test

# Declare external function signatures
from test import add(a: int, b: int) -> int

def main(args:list[str]) {
    result:int = test.add(5, 7)
    print("5 + 7 = ", result, "\n")
}

Importing a .a File


import "./libtest.a" as test

# Declare external function signatures (same syntax as for .so files)
from test import add(a: int, b: int) -> int

def main(args:list[str]) {
    result:int = test.add(5, 7)
    print("5 + 7 = ", result, "\n")
}

14.4 Importing System Libraries


import "libm.so" as math

# Declare external functions
from math import sqrt(x: float) -> float

def main(args:list[str]) {
    result:float = math.sqrt(16.0)
    print("sqrt(16) = ", result, "\n")
}

14.5 Declaring External Functions (from...import Syntax)

When importing .so or .a files, if the compiler cannot automatically recognize the functions in the library, you can use the from...import syntax to manually declare the signatures of external functions. This is similar to Python's from module import func syntax.

Syntax Format


from <lib_alias> import <func_name>(<param_name>: <param_type>, ...) -> <return_type>

Usage Example


# First import the shared library
import "./libmylib.so" as mylib

# Declare external function signatures
from mylib import my_function(a: int, b: int) -> int
from mylib import another_function(s: str) -> int
from mylib import get_value() -> float

def main(args:list[str]) {
    # Call external functions
    result = mylib.my_function(5, 3)
    print("Result: ", result, "\n")
    
    msg = mylib.another_function("hello")
    print("Message: ", msg, "\n")
    
    value = mylib.get_value()
    print("Value: ", value, "\n")
}

Supported Type Mapping

CatBase Type C/Zig Type Description
int c_int Integer
float f64 Floating-point number (C's double)
bool c_int Boolean (typically represented as int in C)
str [*c]const u8 String pointer
bytes [*c]u8 Byte pointer
Pointer [*c]u8 Opaque pointer (input parameter)
*Pointer [*c][*c]u8 Pointer to pointer (output parameter; C function writes a new pointer into the Pointer, e.g. ppDb of sqlite3_open)
None void No return value

*Pointer Output Pointer Parameters

When a C function requires a T** output parameter (the function returns a new pointer through this parameter), use the *Pointer type in the from...import declaration. On the user side, just use a regular Pointer variable (initialized with pointer()); the compiler will automatically:

  1. Declare the Pointer variable as var (not const), so that the C function's write is preserved
  2. Generate an (&var).toPtrPtr() call to get a [*c][*c]u8 passed to the C function
  3. After the call, var's internal ptr field holds the new pointer written by the C function

Typical example (SQLite):


import "/usr/lib/x86_64-linux-gnu/libsqlite3.so" as _sqlite3_native

# ppDb is sqlite3** → declared as *Pointer
from _sqlite3_native import sqlite3_open(filename: bytes, ppDb: Pointer) -> int

def open_db() -> int {
    db_ptr: Pointer = pointer()  # Just a normal Pointer variable
    rc: int = _sqlite3_native.sqlite3_open(path_bytes, db_ptr)
    if rc != 0 {
        return 0
    }
    # Now db_ptr's internal ptr field holds the sqlite3* pointer
    return 1
}

Practical Use Case

When you only have a .so file but not the corresponding header file, you can use this syntax to declare the functions you need to use:


import "/usr/local/lib/libcustom.so" as custom

# Declare the required functions
from custom import process_data(input: bytes, size: int) -> int
from custom import init(config: str) -> int
from custom import cleanup() -> None

def main(args:list[str]) {
    custom.init("debug=true")
    data = bytes("test data")
    result = custom.process_data(data, len(data))
    custom.cleanup()
}

Notes:

  • The from...import statement must come after the corresponding import statement
  • The function name must exactly match the actual function name in the .so file
  • The parameter types and return type must match the C function declaration, otherwise runtime errors may occur

14.6 Official Packages (packages/)

Chapter Overview: CatBase provides a series of official pre-packaged packages in the packages/ directory, which hide all FFI details of common C libraries (SQLite, Opus, etc.) inside the index.catc file. Users only need to import and call functions, with no need to deal with Pointer, libxxx.so paths, from...import or other low-level details.

Design Goals

Official packages follow these design principles:

  1. Zero Low-Level Exposure: User code never needs to write import xxx.so, from xxx import ..., or use Pointer
  2. Python-style API Naming: API naming is consistent with the equivalent Python library, making migration from Python easier
  3. Automatic Resource Management: Connections, handles, buffers, etc. are maintained internally; users only reference them by ID
  4. Cross-platform: All platform-specific code is handled inside index.catc; user code is platform-agnostic

Currently Available Official Packages

Package Path Functionality Python Equivalent
opus packages/opus/index.catc Opus audio codec opuslib
sqlite packages/sqlite/index.catc SQLite embedded database sqlite3
opencv packages/opencv/index.catc OpenCV image processing (C++ bridge) cv2
numpy packages/numpy/index.catc N-dimensional array numerical computing (pure C bridge) numpy

Usage Pattern

Official packages all use packages/<name>/index.catc as the public interface file. The import pattern is uniform:


import <package_name>/index as <alias>

Example 1: Opus Audio Encoding


import opus/index as opus_lib

def main(args: list[str]) {
    enc_id: int = opus_lib.Encoder(48000, 1, opus_lib.APPLICATION_VOIP)
    pcm: bytes = bytes("00" * 960)
    encoded: bytes = opus_lib.encode(enc_id, pcm, 480)
    opus_lib.destroy_encoder(enc_id)
    opus_lib.close()
}

Example 2: SQLite Database


import sqlite/index as sqlite

def main(args: list[str]) {
    conn: int = sqlite.connect(":memory:")
    if conn == 0 {
        print("ERROR: failed to open database\n")
        return
    }

    # Create table
    if sqlite.execute(conn, "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT NOT NULL)") == 0 {
        sqlite.close_all()
        return
    }

    # Insert data
    sqlite.execute(conn, "INSERT INTO users (name) VALUES ('Alice')")

    # Query
    sel: int = sqlite.prepare(conn, "SELECT id, name FROM users")
    while sqlite.step(sel) == sqlite.ROW {
        id_val: int = sqlite.column_int(sel, 0)
        name: str = sqlite.column_text(sel, 1)
        print("id=", id_val, " name=", name, "\n")
    }
    sqlite.finalize(sel)
    sqlite.close_all()
}

Example 3: OpenCV Image Processing

Note: Unlike SQLite/Opus, the OpenCV package requires compiling the C++ bridge layer .so using build.sh first.


import opencv/index as cv

def main(args: list[str]) {
    # 1. Read image
    img: Pointer = cv.imread("photo.jpg", cv.IMREAD_COLOR)
    if !img.is_null() {
        # 2. Convert to grayscale
        gray: Pointer = cv.cvtColor(img, cv.COLOR_BGR2GRAY)

        # 3. Resize to half
        small: Pointer = cv.resize(img, 150, 100, cv.INTER_LINEAR)

        # 4. Query image dimensions
        s: list[int] = cv.shape(img)
        print("size:", s[0], "x", s[1], "x", s[2], "\n")

        # 5. Save
        cv.imwrite("gray.jpg", gray)
        cv.imwrite("small.jpg", small)

        # 6. Release resources (CatBase has no RAII, must call explicitly)
        cv.release_all()
    }
}

OpenCV Package - Structure and Build Process

OpenCV primarily uses the C++ API, while CatBase currently only supports C ABI FFI. Therefore, the OpenCV package uses a different structure than SQLite/Opus:

packages/opencv/
├── src/bridge.cpp          # C++ → C ABI bridge source file
├── lib/libopencv_bridge.so # Compiled output (C ABI shared library)
├── build.sh                # Compiles bridge.cpp using g++
├── setup.sh                # Detects and installs system OpenCV (apt install libopencv-dev)
└── index.catc              # CatBase public API

Usage Steps:

# 1. Install system OpenCV (one-time)
./packages/opencv/setup.sh

# 2. Compile bridge .so (one-time)
./packages/opencv/build.sh

# 3. Write your CatBase code
./bin/catcc examples/opencv_basic.cat
./opencv_basic

Design Rationale:

  1. CatBase → libopencv_bridge.so: CatBase can only call C ABI symbols
  2. libopencv_bridge.so → libopencv_core.so: The C++ bridge layer internally calls the OpenCV C++ API
  3. In the C++ bridge source (src/bridge.cpp), cv::Mat is wrapped as an opaque pointer opencv_mat* to avoid C++ type layout leaking to the C ABI boundary
  4. index.catc provides Python cv2-style high-level APIs (imread / cvtColor / resize etc.), users never touch Pointer or C functions

Example 4: NumPy Numerical Computing

The NumPy package is the core component of CatBase numerical computing, providing a Python numpy-compatible N-dimensional array (ndarray) API for smooth migration of Python developers.


import "numpy/index" as np

def main(args: list[str]) {
    # 1. Create a 3x4 zero array
    s: list[int] = [3, 4]
    a: int = np.zeros(s, np.DTYPE_FLOAT64)

    # 2. Set elements
    np.set(a, 0, 0, 0, 0, 1.0)
    np.set(a, 0, 1, 0, 0, 2.0)
    np.set(a, 1, 0, 0, 0, 3.0)
    np.set(a, 2, 3, 0, 0, 4.0)

    # 3. Query and reduce
    print("a[0,0] =", np.get(a, 0, 0, 0, 0), "\n")
    print("shape =", np.shape(a), "  sum =", np.sum(a), "\n")

    # 4. Array arithmetic
    b: int = np.ones(s, np.DTYPE_FLOAT64)
    c: int = np.add(a, b)
    print("a + b sum =", np.sum(c), "\n")
    np.release(c)
    np.release(b)

    # 5. Release
    np.release(a)
    np.close_all()
}

For complete documentation, see packages/numpy/catbase_numpy_cn.md.

NumPy Package - Structure and Build

The NumPy package uses a pure C implementation (unlike OpenCV's C++ bridge), depending on no external C/C++ libraries — only libc + libm:

packages/numpy/
├── src/numpy_bridge.c         # Pure C ABI implementation (~700 lines)
├── lib/libnumpy_bridge.so     # Compiled output (~26 KB, only depends on libc + libm)
├── build.sh                   # Compile C bridge → .so
├── setup.sh                   # Check gcc/clang environment
├── index.catc                 # CatBase public API (300+ lines, 24 functions)
└── catbase_numpy_cn.md        # Detailed usage manual

Usage steps:

# 1. Check build environment
./packages/numpy/setup.sh

# 2. Build the pure C bridge .so (one-time, very fast ~1 second)
./packages/numpy/build.sh
# Output:
#   /home/.../packages/numpy/lib/libnumpy_bridge.so  (local copy)
#   /usr/local/lib/libnumpy_bridge.so                 (system install, referenced by absolute path in index.catc)

# 3. Write your CatBase code
./bin/catcc examples/numpy/test_numpy_01_basic.cat
./test_numpy_01_basic

Mapping to Python numpy:

Python numpy CatBase numpy Notes
np.zeros([3, 4]) np.zeros([3, 4], np.DTYPE_FLOAT64) dtype must be explicit
np.ones([3, 4]) np.ones([3, 4], np.DTYPE_FLOAT64) dtype must be explicit
arr.shape np.shape(arr) → list[int] property → function
arr[i, j] np.get(arr, i, j, 0, 0) subscript → function
arr[i, j] = v np.set(arr, i, j, 0, 0, v) subscript assignment → function
np.add(a, b) np.add(a, b) identical name
a + 10 np.add_scalar(a, 10.0) operator → function
np.sum(a) np.sum(a) identical name
del arr np.release(arr) auto GC → explicit release

Core features:

  1. Zero external dependencies: Pure C + libc + libm, compiles in ~1 second
  2. Python-style API: Naming and behavior highly consistent with Python numpy
  3. No low-level exposure: User code never needs import xxx.so, from _np import, or Pointer
  4. Reference counting + global registry: Automatic use-after-free prevention, up to 4096 concurrent arrays
  5. 1D-4D support: Row-major storage, consistent with numpy
  6. 7 dtypes: float64/float32/int64/int32/uint8/int8/bool

Currently implemented functions (24):

Category Functions
Creation zeros / ones / full / arange / linspace / copy
Metadata shape / ndim / size / dtype_of / ref_count
Element Access get / set (1-4D universal) / get_1d..4d / set_1d..4d
Reshape reshape (returns view, shares data)
Array Operations add / subtract / multiply / divide / power / maximum / minimum
Scalar Operations add_scalar / sub_scalar / mul_scalar / div_scalar / pow_scalar
Reduction sum / mean / min / max
Resources retain / release / close_all

To extend (matrix multiplication, broadcasting, file I/O, etc.), add C functions in src/numpy_bridge.c and wrappers in index.catc.

SQLite Package API Reference

packages/sqlite/index.catc fully wraps the SQLite C API. Common functions:

Connection Management

  • connect(path: str) -> int: Open/create a database (":memory:" for in-memory database), returns connection ID
  • close_connection(conn_id: int): Close a single connection
  • close_all(): Close all connections
  • last_error(conn_id: int) -> str: Get the most recent SQLite error message (calls sqlite3_errmsg)

Simple Execution (no parameters)

  • execute(conn_id: int, sql: str) -> int: Execute CREATE / INSERT / DELETE / UPDATE etc. (no result set)
  • last_insert_rowid(conn_id: int) -> int: Get the rowid of the most recent INSERT

Parameterized Execution (bound parameters)

  • prepare(conn_id: int, sql: str) -> int: Pre-compile SQL, returns statement ID
  • bind_null(stmt_id, idx): Bind NULL
  • bind_int(stmt_id, idx, value): Bind 32-bit integer
  • bind_int64(stmt_id, idx, value): Bind 64-bit integer
  • bind_double(stmt_id, idx, value): Bind 64-bit float
  • bind_text(stmt_id, idx, str_value): Bind text (auto null-terminated, uses SQLITE_TRANSIENT to copy)
  • run(stmt_id) -> int: Execute the prepared statement
  • step(stmt_id) -> int: Single-step execution (for SELECT, returns SQLITE_ROW / SQLITE_DONE)
  • reset_stmt(stmt_id): Reset the statement so it can be re-executed (without releasing)
  • finalize(stmt_id): Release the statement

Result Reading

  • column_count(stmt_id) -> int: Number of columns
  • column_name(stmt_id, idx) -> str: Column name
  • column_int(stmt_id, idx) -> int: 32-bit integer column
  • column_double(stmt_id, idx) -> float: 64-bit float column
  • column_text(stmt_id, idx) -> str: Text column
  • column_value(stmt_id, idx) -> str: Auto type conversion (int / float / text / null → str)
  • column_type(stmt_id, idx) -> int: Returns SQLITE_INTEGER / SQLITE_FLOAT / SQLITE_TEXT / SQLITE_NULL / SQLITE_BLOB
  • column_is_null(stmt_id, idx) -> bool: Check whether the current column value is NULL

Metadata

  • libversion() -> str: SQLite C library version
  • changes(conn_id) -> int: Number of rows affected by the most recent UPDATE/DELETE
  • SQLITE_OK / SQLITE_ROW / SQLITE_DONE / SQLITE_INTEGER / SQLITE_FLOAT / SQLITE_TEXT / SQLITE_NULL / SQLITE_BLOB: status/type constants

See examples/test_sqlite.cat for a complete example.


15. Audio Recording and Playback

Chapter Overview: This chapter introduces CatBase's audio features, supporting recording and playback. CatBase uses Linux ALSA (Advanced Linux Sound Architecture) for audio processing, providing an API similar to Python PyAudio.

15.1 Dependency Installation

Using audio features on Linux requires installing the ALSA development library:


# Ubuntu/Debian
sudo apt-get install libasound2-dev

# CentOS/RHEL
sudo yum install alsa-lib-devel

15.2 One-shot Recording

record

record(duration:int, sample_rate:int, channels:int, device_name:str, chunk:int) : bytes - Records audio and returns PCM data

Parameters:

  • duration - Recording duration (seconds)
  • sample_rate - Sample rate (e.g. 16000)
  • channels - Number of channels (e.g. 1)
  • device_name - Device name, in the format plughw:CARD=X,DEV=Y, obtainable via getInputDeviceList()
  • chunk - Buffer size (e.g. 1024)

Returns:

  • Returns PCM audio data of type bytes

def main(args:list[str]) {
    print("Starting recording...\n")

    # Record for 5 seconds, return PCM data
    # Parameters: duration, sample_rate, channels, device_name, chunk
    data:bytes = record(5, 16000, 1, "plughw:CARD=0,DEV=0", 1024)

    # Save as WAV file
    save_wav(data, "/tmp/recording.wav", 16000)

    print("Recording complete, saved to /tmp/recording.wav\n")

    # Play the recording (specify output device)
    play(data, 16000, "plughw:CARD=1,DEV=0")

    print("Playback complete\n")
}

15.3 Streaming Recording

recordStream

recordStream(rate:int, channels:int, chunk:int, format:int, device_name:str, callback:function) : RecordStream - Creates a streaming recording object

Parameters:

Parameter Keyword Argument Default Value Description
rate rate 16000 Sample rate (e.g. 16000)
channels channels 1 Number of channels (e.g. 1)
chunk chunk 1024 Buffer size
format format 3 Audio format code
device_name device_name "default" Device name, in the format plughw:CARD=X,DEV=Y
callback callback None Callback function (optional)

Parameter Description:

  • rate - Sample rate, in Hz (e.g. 16000, 44100)
  • channels - Number of channels; 1 means mono, 2 means stereo
  • chunk - Number of audio frames read each time; affects latency and memory usage
  • format - Audio format code, see table below
  • device_name - ALSA device name, obtained via getInputDeviceList()
  • callback - Optional callback function for asynchronous recording

format format codes:

Code Format
1 S8
2 U8
3 S16_LE (default)
4 S16_BE
5 U16_LE
7 S24_LE
11 S32_LE
15 FLOAT
16 FLOAT64
17 MU_LAW
18 A_LAW

RecordStream Methods

Method Return Type Description
read() str Reads one chunk of audio data
is_active() bool Checks whether the recording stream is active
close() None Closes the recording stream
setCallback(callback) None Sets the str callback function (backward compatible)
setBytesCallback(callback) None Sets the bytes callback function (new in v0.0.8)
startRecording() None Starts asynchronous recording with a callback
stopRecording() None Stops recording

Callback Mechanism Description (Enhanced in v0.0.8):

CatBase supports an audio callback mechanism, modeled after the callback pattern of Python sounddevice. When the callback parameter is specified:

  • Recording proceeds asynchronously in a separate thread
  • The callback function is invoked automatically whenever new audio data is available
  • The callback function receives one parameter, supporting both str and bytes types
  • The compiler auto-dispatches: if the callback first parameter is bytessetBytesCallback, otherwise → setCallback

Callback Function Signature:


# Method 1: bytes callback (recommended, works with Queue)
def on_audio_data(data: bytes) {
    audio_queue.put(data, -1)  # ✅ auto-dispatched to fromBytes
}

# Method 2: str callback (backward compatible)
def on_audio_data(data: str) {
    print("Received: ", data, "\n")
}

Recording Example Using a Callback:


audio_queue: Queue = queue(100)

# bytes callback function - invoked when new audio data is available
def on_audio_data(data: bytes) {
    print("Received audio frame\n")
    audio_queue.put(data, -1)  # ✅ auto-uses fromBytes
}

def main(args: list[str]) {
    # Create a recording stream with a callback
    # Parameters: rate, channels, chunk, format, device_name, callback
    stream: RecordStream = recordStream(16000, 1, 512, 3, "plughw:CARD=0,DEV=0", on_audio_data)

    print("Recording started with callback...")
    sleep(5)  # Record for 5 seconds

    stream.stopRecording()
    stream.close()
    print("Recording stopped")
}

Recording Example Using Keyword Arguments:


def on_audio_data(data: bytes) {
    print("Received: ", len(data), " bytes")
}

def main(args: list[str]) {
    # Use keyword arguments
    stream: RecordStream = recordStream(rate=16000, channels=1, chunk=512, format=3, device_name="plughw:CARD=0,DEV=0", callback=on_audio_data)

    print("Recording started...")
    sleep(5)

    stream.stopRecording()
    stream.close()
    print("Recording stopped")
}

Streaming Recording (No Callback) Example:


def main(args:list[str]) {
    # Create a recording stream
    stream: RecordStream = recordStream(44100, 1, 1024, 3, "plughw:CARD=0,DEV=0")

    frames: list[bytes] = []
    data: bytes

    print("Starting recording, press Ctrl+C to stop...\n")

    # Continuously record, checking active state
    while stream.is_active() {
        data = stream.read()
        if len(data) > 0 {
            frames.append(data)
        }
    }

    # Close the stream
    stream.close()

    # Merge all recorded data
    all_data: bytes = bytes("")
    i: int = 0
    while i < frames.len() {
        all_data = all_data + frames[i]
        i = i + 1
    }

    # Save as WAV
    save_wav(all_data, "/tmp/stream_recording.wav", 44100)

    print("Recording complete, saved to /tmp/stream_recording.wav\n")
}

15.4 Playing Audio

CatBase provides two ways to play audio: play and playStream, suitable for different scenarios.

play - One-shot Playback

play(data:bytes, sample_rate:int, device_name:str) : None - Plays PCM audio data

Applicable Scenarios:

  • Playing already-complete audio data (e.g. an entire recording)
  • Simple playback needs that do not require streaming control
  • Playing files (must first be read as PCM data)

Parameters:

  • data - PCM audio data of type bytes
  • sample_rate - Sample rate (e.g. 16000)
  • device_name - Device name, in the format plughw:CARD=X,DEV=Y, obtainable via getOutputDeviceList()

def main(args:list[str]) {
    # Record
    data:bytes = record(3, 16000, 1, "plughw:CARD=0,DEV=0", 1024)

    # Play (one-shot playback of the entire data, specifying output device)
    play(data, 16000, "plughw:CARD=1,DEV=0")
}

playStream - Streaming Playback

playStream(rate:int, channels:int, format:int, device_name:str, callback:function) : PlayStream - Creates a streaming playback object

Applicable Scenarios:

  • Real-time streaming playback needed (e.g. network audio streams, real-time synthesized audio)
  • Need to control playback state (is_active, wait)
  • Need to write audio data in segments
  • Need to play large amounts of audio data over a long period
  • Need precise control over the playback flow

Parameters:

Parameter Keyword Argument Default Value Description
rate rate 16000 Sample rate (e.g. 16000)
channels channels 1 Number of channels (e.g. 1)
format format 3 Audio format code
device_name device_name "default" Device name, in the format plughw:CARD=X,DEV=Y
callback callback None Callback function (optional)

Parameter Description:

  • rate - Sample rate, in Hz
  • channels - Number of channels
  • format - Audio format code, see table below
  • device_name - ALSA device name, obtained via getOutputDeviceList()
  • callback - Optional callback function for asynchronous playback

format format codes:

Code Format
1 S8
2 U8
3 S16_LE (default)
4 S16_BE
7 S24_LE
11 S32_LE
15 FLOAT
16 FLOAT64
17 MU_LAW
18 A_LAW

Comparison of play and playStream:

Feature play playStream
Usage One-shot playback of entire data Create a stream object, write in segments to play
Complexity Simple Slightly more complex
Control capability None Can control playback state, wait for completion
Applicable scenarios Playing short audio, simple needs Real-time streaming playback, large data volume playback
Resource management Automatic release Must manually call close()
Callback support None Supports callback mechanism

PlayStream Methods

Method Return Type Description
write(data) bool Writes audio data
is_active() bool Checks whether the playback stream is active
wait() None Waits for playback to complete
close() None Closes the playback stream
setCallback(callback) None Sets the str callback function (backward compatible)
setBytesCallback(callback) None Sets the bytes callback function (new in v0.0.8)
startPlaying() None Starts asynchronous playback with a callback
stopPlaying() None Stops playback

Playback Callback Mechanism Description:

When the callback parameter is specified, playback proceeds asynchronously in a separate thread:

  • Whenever the audio device needs data, the callback function is invoked automatically to obtain audio data
  • The callback function returns audio data, supporting both str and bytes types
  • The compiler auto-dispatches: if the callback return type is bytessetBytesCallback, otherwise → setCallback
  • When empty data is returned, playback pauses and waits

Callback Function Signature:


# Method 1: bytes callback (recommended)
def get_audio_data() -> bytes {
    return b"\x00\x00\x00\x00"  # Return audio data
}

# Method 2: str callback (backward compatible)
def get_audio_data() -> str {
    return "audio_data"
}

Playback Example Using a Callback:


# bytes playback callback function - invoked when the audio device needs data
def get_audio_data() -> bytes {
    # You can generate or obtain audio data here
    # For example: read from file, synthesize in real time, receive from network, etc.
    return b"\x00\x00\x00\x00"
}

def main(args: list[str]) {
    # Create a playback stream with a callback
    # Parameters: rate, channels, format, device_name, callback
    pstream: PlayStream = playStream(16000, 1, 3, "plughw:CARD=1,DEV=0", get_audio_data)

    print("Playback started with callback...")
    sleep(5)  # Play for 5 seconds

    pstream.stopPlaying()
    pstream.close()
    print("Playback stopped")
}

Playback Example Using Keyword Arguments:


def get_audio_data() -> bytes {
    return bytes("audio_data")
}

def main(args: list[str]) {
    # Use keyword arguments
    pstream: PlayStream = playStream(rate=16000, channels=1, format=3, device_name="plughw:CARD=1,DEV=0", callback=get_audio_data)

    print("Playback started...")
    sleep(5)

    pstream.stopPlaying()
    pstream.close()
    print("Playback stopped")
}

Streaming Playback (No Callback) Example:


def main(args:list[str]) {
    # Create a playback stream
    pstream: PlayStream = playStream(44100, 1, 3, "plughw:CARD=1,DEV=0")

    # Record
    data:bytes = record(5, 44100, 1, "plughw:CARD=0,DEV=0", 1024)

    # Streaming playback
    pstream.write(data)

    # Wait for playback to complete
    pstream.wait()

    # Close the playback stream
    pstream.close()

    print("Playback complete\n")
}

15.5 Audio Device List

CatBase provides the getInputDeviceList() and getOutputDeviceList() functions to obtain system audio device information.

Device Name Format Description

CatBase uniformly uses device names in the plughw:CARD=X,DEV=Y format:

  • plughw: - Uses the ALSA plugin layer, automatically handling sample rate conversion and format conversion; recommended
  • hw: - Direct hardware access, requires the hardware to support the specified format (low latency but may not support some sample rates)

Why use plughw:?

Feature hw:CARD=X,DEV=Y plughw:CARD=X,DEV=Y
Sample rate conversion ❌ Not supported ✅ Automatic conversion
Format conversion ❌ Not supported ✅ Automatic handling
Latency Lower Slightly higher
Applicable scenarios Professional audio General scenarios (recommended)

getInputDeviceList

getInputDeviceList() : list[dict[str, str]] - Gets the list of available input devices (recording devices)

Returns a list of available recording devices; each device is a dict containing the following fields:

Field Type Description
name str Device name, in the format plughw:CARD=X,DEV=Y
description str Device description name

def main(args:list[str]) {
    # Get the input device list
    input_devices:list[dict[str, str]] = getInputDeviceList()
    
    print("Available recording devices:\n")
    i:int = 0
    while i < len(input_devices) {
        dev:dict[str, str] = input_devices[i]
        print("  ", i, ": name=", dev["name"], ", desc=", dev["description"], "\n")
        i = i + 1
    }
    
    # Use the first device to record (skip the 0th default device)
    if len(input_devices) > 1 {
        dev_name:str = input_devices[1]["name"]
        data:bytes = record(3, "16000", "1", dev_name, "1024")
    }
}

getOutputDeviceList

getOutputDeviceList() : list[dict[str, str]] - Gets the list of available output devices (playback devices)

Returns a list of available playback devices; each device is a dict containing the following fields:

Field Type Description
name str Device name, in the format plughw:CARD=X,DEV=Y
description str Device description name

def main(args:list[str]) {
    # Get the output device list
    output_devices:list[dict[str, str]] = getOutputDeviceList()
    
    print("Available playback devices:\n")
    i:int = 0
    while i < len(output_devices) {
        dev:dict[str, str] = output_devices[i]
        print("  ", i, ": name=", dev["name"], ", desc=", dev["description"], "\n")
        i = i + 1
    }
    
    # Use the first device to play (skip the 0th default device)
    if len(output_devices) > 1 {
        dev_name:str = output_devices[1]["name"]
        data:bytes = record(3, 16000, 1, "default", 1024)
        play(data, 16000, dev_name)
    }
}

15.6 Saving as a WAV File

save_wav

save_wav(data:str, filename:str, sample_rate:str) : None - Saves PCM data as a WAV file

Parameters:

  • data - PCM audio data of type bytes
  • filename - File path to save
  • sample_rate - Sample rate string (e.g. "16000")

WAV File Format:

  • Sample rate: configurable (default 16000 Hz)
  • Channels: 1 (mono)
  • Bit depth: 16-bit
  • Format: PCM

def main(args:list[str]) {
    data:bytes = record(5, 16000, 1, "plughw:CARD=0,DEV=0", 1024)
    save_wav(data, "/tmp/my_recording.wav", 16000)
    print("Saved as WAV file\n")
}

15.7 Complete Examples

Example 1: Simple Recording and Playback (Using the Device List)


def main(args:list[str]) {
    # Get device lists
    input_devices:list[dict[str, str]] = getInputDeviceList()
    output_devices:list[dict[str, str]] = getOutputDeviceList()
    
    # Skip the 0th default device, use the first actual device
    if len(input_devices) < 2 or len(output_devices) < 2 {
        print("No audio devices found\n")
        return
    }
    
    in_dev:str = input_devices[1]["name"]
    out_dev:str = output_devices[1]["name"]
    
    print("Starting recording...\n")

    # Record for 5 seconds, return PCM data
    data:bytes = record(5, 16000, 1, in_dev, 1024)

    # Save as WAV file
    save_wav(data, "/tmp/recording.wav", 16000)

    print("Recording complete, saved to /tmp/recording.wav\n")

    # Play the recording
    play(data, 16000, out_dev)

    print("Playback complete\n")
}

Example 2: Using Streaming Recording and Playback


def main(args:list[str]) {
    # Get the device list
    input_devices:list[dict[str, str]] = getInputDeviceList()
    
    if len(input_devices) < 2 {
        print("No recording devices found\n")
        return
    }
    
    in_dev:str = input_devices[1]["name"]
    
    # Use one-shot recording
    frames: list[bytes] = []
    
    print("Starting recording...\n")
    
    # Record 5 times, 1 second each
    i: int = 0
    while i < 5 {
        data:bytes = record(1, 16000, 1, in_dev, 1024)
        frames.append(data)
        i = i + 1
    }
    
    # Merge data
    all_data: str = ""
    i = 0
    while i < frames.len() {
        all_data = all_data + frames[i]
        i = i + 1
    }
    
    save_wav(all_data, "/tmp/recording.wav", 16000)
    print("Done!\n")
}

Example 3: Multi-segment Recording


def main(args:list[str]) {
    # Get the device list
    input_devices:list[dict[str, str]] = getInputDeviceList()
    
    if len(input_devices) < 2 {
        print("No recording devices found\n")
        return
    }
    
    in_dev:str = input_devices[1]["name"]
    
    # Record 10 segments, 1 second each
    frames: list[bytes] = []
    segment_count: int = 10
    
    i: int = 0
    while i < segment_count {
        print("Recording segment")
        print(i + 1)
        print("...\n")

        data:bytes = record(1, 16000, 1, in_dev, 1024)
        frames.append(data)

        i = i + 1
    }
    
    # Merge and save
    all_data: str = ""
    j: int = 0
    while j < frames.len() {
        all_data = all_data + frames[j]
        j = j + 1
    }
    
    save_wav(all_data, "/tmp/segments.wav", 16000)
    print("Done!\n")
}

15.8 Comparison with Python PyAudio

Recording Stream:

PyAudio CatBase Description
pyaudio.PyAudio() Built-in No need to create
stream = p.open(format=FORMAT, channels=CHANNELS, rate=RATE, input=True, frames_per_buffer=CHUNK) stream: RecordStream = recordStream(rate=RATE, channels=CHANNELS, chunk=CHUNK, format=FORMAT) Create recording stream
stream.read(CHUNK) stream.read() Read audio data
stream.is_active() stream.is_active() Check active state
stream.stop_stream() stream.close() Stop/close stream

Playback Stream:

PyAudio CatBase Description
stream = p.open(format=FORMAT, channels=CHANNELS, rate=RATE, output=True, frames_per_buffer=CHUNK) pstream: PlayStream = playStream(rate=RATE, channels=CHANNELS, format=FORMAT) Create playback stream
stream.write(data) pstream.write(data) Write audio data
stream.is_active() pstream.is_active() Check playback state
stream.stop_stream() pstream.close() Stop/close stream

16. Syntax Summary

Chapter Overview: Through the previous chapters, you have mastered all the core features of CatBase. This chapter summarizes all the syntax for easy reference and review. Through this syntax summary, you can quickly revisit CatBase's keywords, data types, built-in functions, and other concepts.

16.1 Keywords

Keyword Description
def Define a function
if Conditional statement
else Otherwise
for Loop (supports lists, ranges, iterators)
while Loop
return Return value
break Break out of a loop
try Try block
except Catch exception
catch Catch exception (compatible)
finally Finally block (always executes)
Exception Exception type
thread Create a thread
async Define a coroutine function
await Wait for a coroutine
import Import a module
from Import from a module (used to declare external functions)
as Alias
None Null value
True Boolean true
False Boolean false
in Membership test / for-in iteration
is Identity test
not Logical not
and Logical and
or Logical or

16.2 Data Types

Type Description
int Integer
float Floating-point number
str String
bool Boolean
list List
dict Dictionary
bytes Byte sequence
function Function reference
None Null value type
any Any type (used for external function declarations)

16.3 Built-in Functions Summary

Function Description
Printing/Input
print(...) Print
input(prompt) Read input
Type Conversion
int(x) Convert to integer
float(x) Convert to floating-point
str(x) Convert to string
bool(x) Convert to boolean
list(x) Convert to list
dict(x) Convert to dictionary
bytes(x) Convert to byte sequence
Type Checking
type(x) Get type
isinstance(x, type) Type check
assert(cond, msg) Assert
Math Operations
abs(x) Absolute value
max(a, b) Maximum
min(a, b) Minimum
pow(a, b) Exponentiation
round(x) Rounding (returns the same type as input)
round(x, n) Round keeping n decimal places (returns float)
sum(list) Summation
Base Conversion
bin(x) Convert to binary
oct(x) Convert to octal
hex(x) Convert to hexadecimal
Character Conversion
chr(x) Integer to character
ord(x) Character to integer
bytes Operations
bytes_alloc(n) Smart allocation (≤4096 static pool, >4096 heap)
bytes_free(b) Smart deallocation (auto-detects source)
bytes_len(b) Get bytes length
bytes_ptr(b) Get bytes data pointer (Pointer type)
bytes_from_array(arr) Create bytes from byte[N] array (copy)
String/Container Operations
len(x) Get length
range(start, end, [step]) Generate range
append(list, item) Append element to list
pop(list, [index]) Pop element
insert(list, index, item) Insert element
remove(list, item) Remove element
File Operations
file(filename, mode) Open file
close(f) Close file
Network Programming
tcpsocket() Create a TCP socket
udpsocket() Create a UDP socket
http_get(url, timeout) HTTP GET
http_post(...) HTTP POST
websocket(url, headers[optional]) WebSocket connection
JSON Processing
json_dumps(dict) Dictionary to JSON
json_loads(str) JSON to dictionary
Serial Communication
serial(port, baud_rate) Open serial port
Threads/Synchronization
thread func(args) Create a thread (statement)
thread_join(t) Wait for thread to finish
mutex() Create a mutex lock
lock(m) Lock
unlock(m) Unlock
queue() Create a message queue
Time Functions
time() Get current timestamp
perf_counter() High-resolution timer
strftime(format, timestamp) Time formatting
Audio
record(seconds, rate, ...) Recording
play(data, rate, device) Play audio
save_wav(data, filename, ...) Save as WAV file
recordStream(...) Create a recording stream
playStream(...) Create a playback stream
getInputDeviceList() Get input device list
getOutputDeviceList() Get output device list
Pointers
pointer(addr) Create a pointer
pointer_of(var) Get a variable's pointer (smart dispatch: bytes returns data pointer, others return variable address)
Others
sleep(seconds) Sleep
exec(code) Execute CatBase code

Afterword

About the Author

The CatBase programming language was designed and developed by Bell Zhong from China. Bell Zhong is also the founder of DoRobot Technology (http://dorobot.net) , a company based in Zhuhai, China.

DoRobot Technology is an innovative company focused on artificial intelligence and robotics, dedicated to developing intelligent solutions. As an in-house programming language, CatBase was originally created to address efficiency and performance issues encountered during project development. The CatBase programming language was born for AI application research and development; it can replace C in minimal runtime environments and supports Python syntax.

Acknowledgments

Thanks to the following people for their support and contributions to CatBase:

  • All CatBase language enthusiasts - Thank you for your choice and trust
  • Open-source community -

                  Thanks to the Zig language team for creating such an excellent compiler toolchain
                  Thanks to the Python community for providing syntax inspiration and reference for CatBase
                  Thanks to the C language ecosystem for providing performance optimization reference for CatBase
                  Thanks to the CatBase community for providing feedback and suggestions
                  Thanks to DoRobot Technology Co., Ltd. from Zhuhai, Guangdong, China (http://dorobot.net) for providing financial support

Future Outlook

CatBase will continue to iterate and improve. Future plans include support for:

  • More built-in data types
  • A richer standard library
  • Cross-platform support (Windows, macOS, etc.)
  • Better IDE support

We believe that CatBase will become a practical, efficient, and easy-to-learn programming language, helping more developers realize their ideas.

Contact Us

  • Official website: http://catbase-lang.com

CatBase Programming Language Reference Manual - Version 1.0