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-staticand-sharedat the same time - Conflicts with .so/.a imports: if the code uses
importto 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:
- Use dynamic linking (default): the target machine needs the MySQL client library installed
``bash catcc source.cat ``
- 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 codelibsource.so- shared library (generated when the-sharedparameter 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
- Create a
catbasefolder in the extensions directory:- Windows:
%USERPROFILE%\.vscode\extensions\catbase - Mac/Linux:
~/.vscode/extensions/catbase
- Windows:
- Create the following file structure in that folder:
catbase/
├── package.json
├── language-configuration.json
└── syntaxes/
└── catbase.tmLanguage
- 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*\()'
- 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.
- 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.
- 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\ ``
- Install the VSIX:
- Open VS Code / Trae
- Press
Ctrl+Shift+Xto open the Extensions view - Click the
...menu in the upper-right corner - Select "Install from VSIX..."
- Select the downloaded
catbase-language-1.0.0.vsixfile - After installation, press
Ctrl+Shift+P, typeDeveloper: Reload Windowto 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:
- Automatic compatibility with
int:byteandintcan be assigned to each other without explicit conversion - Constructor
byte(x): truncates or wraps an int/float into the 0-255 range - Accepts
None:byte(None) = 0 - Convert to a single character via
chr():chr(byte(65)) = "A" - Range limit: values outside 0-255 will be truncated/wrapped by Zig's
@intCast - Single-character bytes literal
b"X":b"A"auto-promotes tobyte(65)- nobyte()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:
| Feature | byte | int |
|---|---|---|
| Zig type | u8 (8-bit) | i64 (64-bit) |
| Range | 0 - 255 | -9.2×1018 ~ 9.2×1018 |
| Memory | 1 byte | 8 bytes |
| chr compatible | ✅ chr(b) = "char" | ✅ chr(65) = "A" |
| Use cases | Bytes, ASCII codes, raw data | General 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:
- Need to be created using built-in functions or constructors
- Usually need to manually release resources (call the
close()method) - Support specific method calls, such as
stream.read(),ws.send(),client.recv(), etc. - 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 ondatatype bytestype is cloned and owned by the queue (avoids data being freed when thread exits)getBytes()transfers ownership; the returned[]u8must 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_msparameter ofput(item, timeout_ms):-1means wait indefinitely,0means non-blocking, a positive number means the number of milliseconds to waitjoin()waits for all messages in the queue to be processed (unfinished_tasks == 0) and for all child threads started viathreadto 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:
- Global variables must be declared outside functions
- Global variables must be initialized at declaration
- Global variables can be accessed and modified in any function
- Global variables retain their values throughout program execution
- Global variables do not support the
varkeyword; 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:
| Form | Signature | Use case | Receives command-line args |
|---|---|---|---|
| Full form | def main(args: list[str]) { } | Programs that need to read user-provided command-line arguments | ✅ Yes |
| Simplified form | def 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
mainfunction) - The simplified form does not error, only warns; it does not leak any low-level language details
- If the simplified
mainbody 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
| Scenario | Recommended form |
|---|---|
| CLI tool (parses user arguments) | Full form def main(args: list[str]) |
| GUI program / background service / daemon | Simplified form def main() |
| Teaching example / Hello World | Simplified form def main() |
| One-shot script / file-processing tool | Simplified form def main() |
Library / framework demo() function | Simplified 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 toi64by default) andfloat(corresponds tof64by 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 intis equivalent toi64, andfloatis equivalent tof64
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 stringsf'...{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_freeauto-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 API—pointer_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 asbytes - ⚠️
b""(empty) isbytestype, notbyte - ✅ 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:
anytype variables can be assigned values of different types multiple times- The
type()function returns anany:actual_typeformat foranytype 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 ananyvalue anytype variables can be assignedNoneto 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
anytype variable, its declared type remainsanyand does not change to the actual value's type - Before performing arithmetic operations on
anytype variables, use type conversion functions (such asint(),float()) to extract the concrete value first dict.get(key)returnsNonefordict[str, any]when the key is not found; for specific types likedict[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:
| Keyword | Meaning | Example |
|---|---|---|
is int | Is it an integer? | if x is int { ... } |
is float | Is it a float? | if x is float { ... } |
is str | Is it a string? | if x is str { ... } |
is bool | Is it a boolean? | if x is bool { ... } |
is list | Is it a list? | if x is list { ... } |
is dict | Is it a dict? | if x is dict { ... } |
is bytes | Is it a byte sequence? | if x is bytes { ... } |
is None | Is it None? | if x is None { ... } |
is not T | Is 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:
| Scenario | Recommended Approach |
|---|---|
any is already a known type but the compiler can't tell | Direct conversion: str(x) / int(x) |
any may be one of several types and requires type-branched handling | Guard: 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
| Need | Syntax | Notes |
|---|---|---|
Declare an any variable | x: any = 42 | Explicitly declare the type as any |
Store None | x: any = None | Any type can hold None |
| Check type | if x is T { ... } | T = int / str / float / bool / list / dict / bytes / None |
| Reverse check | if x is not None { ... } | Exclude a type |
| Use after guard | if x is int { n := x; ... } | x is narrowed to int in the then block |
Convert explicitly to str | s: str = str(x) | No guard required; direct conversion |
Convert explicitly to int | n: int = int(x) | Without a guard, non-int values return 0 |
Convert to float | f: float = float(x) | Same as above |
| Get type explicitly | type(x) | Returns the "any:actual_type" string |
| Container nesting | dict[str, any] | Standard pattern for JSON data |
| Mixed-type list elements | list[any] | List of any type |
| Multi-level nesting | dict[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
| Keyword | Style | Target Audience |
|---|---|---|
struct | C / Zig / Rust style, emphasizes "value-type aggregate" | Systems programming background (Rust/Go/C programmers) |
class | Python / 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):
- Prescan stage: when the compiler sees a method call (e.g.
add/append/put) on a struct field such asv.key, it automatically promotes the struct variablevtovar. Reason: in Zig, aconstfield is read-only, so methods taking a*Treceiver are rejected. - codegen stage: the receiver (e.g.
v.key) is automatically wrapped with@constCast(&v.key), sov.key.add()compiles. Chained calls such asv.nested.field.method()are also supported. - 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
corc.fooas variable names. CatBase treatsc.xxxas a C library function call (e.g.c.malloc); using it as a struct field will fail to compile. Prefer semantically named variables such asbox/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:
| Scenario | What it verifies | Example |
|---|---|---|
0-arg set() initialization | LHS context propagates through fields | box: Empty = Empty { items: set(), tags: set() } |
| Full set API on struct fields | add / len / contains / unionSet / intersection / difference | alice.friends.contains("bob"), alice.friends.unionSet(other.friends) |
| 3-level nesting (struct > list[struct] > set) | chained access + method calls on list-element struct fields | outer.middle.items[0].tags.add("a-new") |
Mixed nesting (struct field is dict[str, set[int]]) | composition of multiple container types | cache.entries: {"session-1": set([1,2,3])} |
| Large set + dedup | 50 elements added twice each still results in 50 | while 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 allowslist[Struct]as an intermediate layer to hold multiple instances.
Complete set method list (compatible with Python set):
| Method | Signature | Description | Python equivalent |
|---|---|---|---|
add(x) | set[T] → bool | Add an element | set.add() |
remove(x) | set[T] → bool | Remove an element | set.remove() |
contains(x) | set[T] → bool | Membership check | x in s |
len() | set[T] → int | Element count | len(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] | Intersection | s & other |
difference(other) | set[T] → set[T] | Difference | s - 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:
| Language | Similar Concept | Inheritance? |
|---|---|---|
| C# | struct (value type) | ❌ No |
| Rust | struct + impl block | ❌ (uses trait) |
| Swift | struct (value type) | ❌ No |
| Kotlin | data class | ❌ (final by default) |
| Go | type T struct { ... } | ❌ No |
| Zig | const 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:
| Year | Language | Inheritance Introduction |
|---|---|---|
| 1967 | Simula 67 | First time "class" and inheritance were introduced together into a programming language |
| 1972 | Smalltalk | Complete OOP system (class + inheritance + dynamic dispatch) |
| 1983 | C++ | Multiple inheritance + virtual function table (vtable) |
| 1995 | Java | Single inheritance + multiple interface implementation |
| 2000 | C# | Single inheritance + interface |
| 2011 | Kotlin | open keyword controls inheritability (final by default) |
| 2014 | Swift | Single inheritance + protocol |
Comparison with other languages' inheritance:
| Language | Inheritance Model | vtable | RTTI | Multiple | Default Inheritable |
|---|---|---|---|---|---|
| C++ | Single/Multiple | ✅ | ✅ | ✅ | ✅ |
| Java | Single | ✅ | ✅ | ❌ (uses interface) | ❌ (final by default) |
| C# | Single | ✅ | ✅ | ❌ (uses interface) | ✅ (unless sealed) |
| Python | Multiple | ✅ (MRO) | ✅ | ✅ | ✅ |
| Rust | No class inheritance | N/A | ❌ | ❌ (uses trait) | N/A |
| Zig | No inheritance | ❌ | ❌ | ❌ | N/A |
| Go | No class inheritance | ❌ (uses interface) | ❌ | ❌ | N/A |
| CatBase | No inheritance | ❌ | ❌ | ❌ | N/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.nameis more explicit than inheriteddog.name(though more verbose) - ✅ Avoids the diamond inheritance problem: Python/Java/C++'s diamond problem simply cannot exist
- ✅ Easier to test: you can mock
Animalalone, without depending on an inheritance hierarchy
3.4.7 Practical Project Recommendations
| Project Type | Recommended Keyword | Reason |
|---|---|---|
| New project (for Python/JS users) | class | Lowers the entry barrier |
| New project (for systems programmers) | struct | Consistent with Zig/Rust style |
| Cross-FFI project | struct | Consistent with C ABI naming |
| Mixed team | Pick one and stick with it | Avoid inconsistent style within the team |
Core principles:
- Use only one keyword within the same project (don't mix, to avoid confusing readers)
- Use
structfor types that cross FFI (aligns with C ABI) - Use
classfor 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:
| Dimension | struct | class | Same? |
|---|---|---|---|
| Number of fields | Determined by definition | Determined by definition | ✅ Identical |
| Memory size | Known at compile time | Known at compile time | ✅ Identical |
| C ABI layout | struct | struct | ✅ 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:
Noneis CatBase's "no value" marker, similar to Python'sNone, C'sNULL, or Java'snull. It is a singleton value representing "no value" or "value unknown". This section explains how to useNone, its type conversion rules, and how it interacts with theanytype anddict.get().
3.5.1 Core Properties of None
Noneis a value, not a type: LikeTrueandFalse,Noneis a concrete value that can be assigned to any variable whose type supportsNone.- Can be assigned to any type:
int,str,float,bool,byte,bytes,any,list[T], anddict[K, V]can all be initialized withNoneto mean "this variable currently has no value". - Unified null check: Use
is None/is not Noneto test whether a variable is empty. - Most useful with the
anytype: Theanytype can storeNoneand 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):
| Conversion | Result | Description |
|---|---|---|
int(None) | 0 | Empty value treated as integer 0 |
float(None) | 0.0 | Empty value treated as float 0.0 |
str(None) | "None" | String form (same as Python str(None)) |
bool(None) | False | Empty value treated as false (same as Python) |
bytes(None) | empty bytes | Byte sequence of length 0 |
byte(None) | 0 | Single 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 isany): returnsNonewhen not founddict[str, T](T is a concrete type such asint/str): returns the zero value of T (0/""/False), notNone
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)returnsany:Nonefor ananyvariable that holdsNone- Assigning
Noneto ananyvariable means "no value" - When parsing JSON,
nullis automatically converted toNone
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
| Note | Description |
|---|---|
Use is None / is not None | Do 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 errors | Before 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:
| Scenario | Use instead |
|---|---|
| Dictionary parsed from JSON | dict[str, any] (concrete container type, value can still be any) |
| Dict elements inside a list | dict[str, any] (same as above) |
| Scalar values inside a list | int / str / float / bool |
| Custom structured data | struct 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
| Trigger | Compiler 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.field | Soft 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
- Prefer
dict[str, any]overanywhen working with JSON / config / parsed data. - Make list element types concrete:
list[dict[str, any]]is more readable and produces fewer errors thanlist[any]. - If you really need a temporary
anyvariable, only do assignment,str()/int()conversion, oris Nonechecks on it — do not call methods. - Fix
[any-usage]warnings immediately: they almost always indicate a more specific type is appropriate. [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:
| Style | Applicable Scenario | Advantage | Disadvantage |
|---|---|---|---|
if x is T { use(x) } | Type of x is uncertain; needs branching | Type-safe, highly readable | Verbose |
str(x) / int(x) | x is already a known type | One-liner | Returns default value (0 / empty string) for unexpected types |
Practical Experience:
- JSON / config parsing → always use
isguard first (value types are truly uncertain) - Intermediate variables inside functions → use
int()/str()directly (avoid unnecessary if nesting) - Return value of
dict.get()→ useisguard before use - Iterating mixed-type list → must use
isguard 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):
()- Parenthesesnot- Logical NOT* / %- Multiplication, division, modulo+ -- Addition, subtraction< > <= >=- Comparison== !=- Equal to / not equal toin- Membership operatorand- Logical ANDor- 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:
- The object implements the
iter_lines()method - The method returns an iterator that yields one line of content per iteration
- 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=Trueparameter 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 exceptionresponse.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:
- Use
startswith("data: ")to check whether it is a data line - Use slice
[6:]to extract the JSON string - Use
json_loads()to parse JSON into a dictionary - Use the
inoperator to check whether a key exists - Use the
.get()method to safely retrieve nested values - Use
str()to convert Variant to a string - 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
continuekeyword. To skip certain iterations, use anifconditional 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:
- No-variable syntax (recommended):
except { ... }- no need to specify an exception variable - Variable syntax:
except errname- capture the exception into a variable - Typed syntax:
except Exception as errname- capture the exception into a variable
Note: The legacy
catchkeyword is also compatible, with the same usage asexcept.
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
- Unified exception type: All exceptions are of type
Exception; there is no distinction between error and Exception - Exception variable scope: The exception variable (such as
eorerr) is only visible inside the except block - Catch anywhere: try-except can be used anywhere in a function
- Keyword choice:
exceptis recommended, butcatchis 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 aslist[int]) based on the actual list literal or variable type returned in the return statement. Similarly, when a variable is declared aslistbut 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
functiontype - 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 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:
| Input | chr(x) / ord(x) | str(x) / int(x) |
|---|---|---|
| 65 | chr(65) = "A" (1 char) | str(65) = "65" (2 chars) |
| 10 | chr(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 fromstr(int)). For a "numeric string" form, convert tointfirst: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
stris 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 stringstr[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:
assertaccepts 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 |
|---|---|
int | i64 |
float | f64 |
bool | bool |
str | runtime.Str |
byte | u8 |
i8/i16/i32/i64 | i8/i16/i32/i64 |
u8/u16/u32/u64 | u8/u16/u32/u64 |
f32/f64 | f32/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 beint(CatBase's int type maps to i64 at the underlying level) - Calling
get()orset()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
- Linux/macOS: such as
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 namemode- 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).
| Mode | Can 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 (0means "fromoffsetto end of file")
Platform Implementation:
- POSIX (Linux/macOS): via
fcntl(F_SETLK)/fcntl(F_SETLKW)system calls - Windows: via
LockFileEx/UnlockFileWin32 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
-
Advisory lock
CatBase's file region locks are advisory. The OS does not enforce them; all collaborators must call
lock/tryLockexplicitly to prevent conflicts. -
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.
-
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 -
Locks are released when the file is closed
f.lock(0, 10) f.close() # automatically releases the [0, 10) lock -
The file must be opened in "r+" mode
lock/tryLock/unlockneed a writable file descriptor. If you open with"r"(read-only),fcntlwill returnEBADFandtryLockwill always returnfalse.f:File = file("data.txt", "r+") # correct f:File = file("data.txt", "r") # wrong — locking will fail -
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:
- Non-blocking
tryLock - Different regions can be locked simultaneously
- Safe write while holding the lock
- Re-locking after
unlock - 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/containsoperations, 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/containsare all constant-time - Set algebra:
unionSet(union) /intersection/difference/symmetricDifference - Subset checks:
isSubset/isSuperset/equals/intersects - Automatic deduplication: when a
setis created from alist, duplicate elements are silently dropped - Interoperable with
list:toListconverts a set to a list
Comparison with list:
list[T]: ordered, allows duplicates — use when order matters or duplicates are allowedset[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]andset[str]. Other element types (such asset[float]orset[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):
| CatBase | Python | Source 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,
settypes still uses Zig's default HashMap debug format (e.g.{"1":"void", ...}) rather than the CatBase-style set output. The compile-time type inference forset()0-arg / 1-arg is fully working; only the nested print formatting needs further polish.
| Function | Description | Recommendation |
|---|---|---|
set() 0-arg | T inferred from LHS, returns empty set | ⭐⭐⭐ Recommended |
set([...]) 1-arg | Build 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 form | Deprecated |
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, notunionThis is because
unionis a reserved keyword in Zig (it is used to declare tagged union types). Naming the methodunionwould clash with that keyword and fail to compile. CatBase intentionally usesunionSetto 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
-
Uniqueness
Sets automatically deduplicate. Calling
addwith an element that is already present has no effect. -
Order is not preserved
Sets are unordered. The list produced by
toListfollows the hash function's internal order, do not depend on any particular ordering. -
No index access
Sets do not support
s[0]-style indexing, because they are unordered. If you need indexed access, convert to a list withtoListfirst. -
No direct iteration (current version)
CatBase does not yet provide
for x in ssyntax for sets. To iterate, usetoList:s:set[int] = set_from_list_int([1, 2, 3]) for x in s.toList() { print(x) } -
Printing a set directly has limitations (to be improved)
Calling
print(s)prints a representation of the form{...}, but using an f-string likeprint(f"{s}")only prints the literal{s}instead of the actual contents. Preferprint(s)overprint(f"...{s}..."). -
Element type is fixed
set[int]andset[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 -
Union method is named
unionSet, notunionBecause
unionis 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:
- Basic
set[int]operations (add/remove/contains/len/clear/isEmpty) - Basic
set[str]operations - Build from a list (
set_from_list_int/set_from_list_str) - Union (
unionSet) - Intersection (
intersection) - Difference (
difference) - Symmetric difference (
symmetricDifference) - Subset / superset checks (
isSubset/isSuperset/equals) - Intersection check (
intersects) - Convert to list (
toList) - Composite example: list deduplication
- 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:
| Container | Order sensitivity | Notes |
|---|---|---|
list[X] | Order-sensitive | Combine in insertion order; [1,2,3] ≠ [3,2,1] |
set[X] | Order-insensitive | Internal sort then combine; {1,2,3} = {3,2,1} |
dict[K,V] | Order-insensitive | Sort by key then combine; {"a":1,"b":2} = {"b":2,"a":1} |
list[list[X]] | Outer + inner list both sensitive | Every list layer combines by position |
list[set[X]] | Outer sensitive + inner set insensitive | Nested sets auto-sort internally |
list[dict[K,V]] | Outer sensitive + inner dict insensitive | Nested 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/dictinto a 64-bit int usable as aDictkey (CatBase'sDict[K,V]requires hashable keys;intis 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
equalstraversal 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])emitsruntime.hashListInt(...);hash(set([1,2,3]))emitsruntime.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_listfirst). - Does not support deeper nesting such as
set[set[X]]/dict[set[X]]/set[dict[K,V]](only thelist[nested]layer is supported); for deeper nesting, manually applyhash(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
jsonparameter is passed, automatically set toapplication/jsonand serialize the dictionary to JSON - If the
dataparameter is passed, automatically set toapplication/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-299iter_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", "")returnsstrtype.get("count", 0)returnsinttype.get("ratio", 0.5)returnsfloattype.get("enabled", True)returnsbooltype.get("data", {})returnsdict[str, any]type.get("items", [])returnslist[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
WebSockettype 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:
- Ensure that only one thread at a time can access shared resources
- Prevent data races: When one thread is modifying data, other threads must wait
- 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:
- When accessing shared variables in a multithreaded environment, you must use a Mutex for protection
- The scope of the lock should be as small as possible, only protecting the necessary critical section code
- 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 timeouttimeout_ms = -1: Wait indefinitelytimeout_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:
- Producer-consumer pattern: One thread produces data, another thread consumes data
- Task dispatching: The main thread dispatches tasks to worker threads
- Event notification: Passing events or messages between threads
- 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:
- Notify all worker threads of the shutdown request
- Worker threads check the shutdown status and stop in an orderly manner
- 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.QueueAPI.
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
Queuetype 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_msvalues: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
ShutdownEventmechanism, 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
- SIGKILL cannot be caught: When a process receives the
SIGKILLsignal, the operating system immediately terminates the process; no cleanup code can be executed
- 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 checkshutdown_requested()and exit in an orderly manner
- Check frequency: Worker threads should periodically check
shutdown_requested(); it is recommended to check on every loop iteration or at fixed time intervals
- 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
Configtype to read and parse INI-format configuration files, similar to Python'sconfigparser. 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 defandawaitsyntax 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 defdefines a coroutine function- Inside a coroutine function, the
awaitkeyword 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:
awaitcan only be used inside anasync deffunctionawaitpauses 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 defexecute in a separate thread - An
awaitcall creates a new thread and joins it to wait for completion async/awaitpreserves 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:
- Importing .cat files: import other CatBase source files and call the functions defined in them
- Importing .catc files: import package files (functionally identical to .cat, but indicates "this is a package to be imported by others")
- Importing .so files: import shared libraries and call C functions
- 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
importstatement may reference only 1 target file; for example,import a bis 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. theaudio.subform); 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):
index.catcfirst: replace.with/and append/index.catcto the last segment as the package entry file- for example,
audio.sub→packages/audio/sub/index.catc
- for example,
libconfig.conffallback: if noindex.catcexists in that directory, look uplibconfig.confin the package root and locate an entry by its<alias>|<target>|<distro>three-part section name (see below)- Last-segment
.sodirect load: if the package path's last segment is literally.so(including versioned forms likelibc.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.so→packages/audio/test.so - for example,
libc.so.6→packages/libc.so.6
- for example,
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 whenpackages/audio/index.catcis 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 theimportalias, target triple (e.g.x86_64-linux-gnu), and distro (e.g.ubuntu,debian,raspbian) path: the actual library file path used at compile timesoname: the SONAME used at link / run time (passed todlopen)
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/.catcpackage files: the suffix is mandatory (explicit suffixes are recommended to avoid ambiguity) - Referencing
.so/.alibrary files: the suffix is mandatory
The wildcard
*has been completely removed fromimportstatements 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 topackages/<X>/<Y>/<Z>/but that directory has neitherindex.catcnor aZalias in any ancestor'slibconfig.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 Macriscv64-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=anddistro=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.confmatches the current target/distro pair; either add a corresponding section or switch the-target/-distroflags.
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 canimportthe same package on every target without any code change. Developers only need to switch-target/-distroto 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:
- Declare the
Pointervariable asvar(notconst), so that the C function's write is preserved - Generate an
(&var).toPtrPtr()call to get a[*c][*c]u8passed to the C function - After the call,
var's internalptrfield 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...importstatement must come after the correspondingimportstatement - 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 theindex.catcfile. Users only need toimportand 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:
- Zero Low-Level Exposure: User code never needs to write
importxxx.so,from xxx import ..., or use Pointer - Python-style API Naming: API naming is consistent with the equivalent Python library, making migration from Python easier
- Automatic Resource Management: Connections, handles, buffers, etc. are maintained internally; users only reference them by ID
- 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:
- CatBase → libopencv_bridge.so: CatBase can only call C ABI symbols
- libopencv_bridge.so → libopencv_core.so: The C++ bridge layer internally calls the OpenCV C++ API
- In the C++ bridge source (
src/bridge.cpp),cv::Matis wrapped as an opaque pointeropencv_mat*to avoid C++ type layout leaking to the C ABI boundary index.catcprovides Python cv2-style high-level APIs (imread/cvtColor/resizeetc.), 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:
- Zero external dependencies: Pure C + libc + libm, compiles in ~1 second
- Python-style API: Naming and behavior highly consistent with Python numpy
- No low-level exposure: User code never needs
importxxx.so,from _np import, or Pointer - Reference counting + global registry: Automatic use-after-free prevention, up to 4096 concurrent arrays
- 1D-4D support: Row-major storage, consistent with numpy
- 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 IDclose_connection(conn_id: int): Close a single connectionclose_all(): Close all connectionslast_error(conn_id: int) -> str: Get the most recent SQLite error message (callssqlite3_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 IDbind_null(stmt_id, idx): Bind NULLbind_int(stmt_id, idx, value): Bind 32-bit integerbind_int64(stmt_id, idx, value): Bind 64-bit integerbind_double(stmt_id, idx, value): Bind 64-bit floatbind_text(stmt_id, idx, str_value): Bind text (auto null-terminated, uses SQLITE_TRANSIENT to copy)run(stmt_id) -> int: Execute the prepared statementstep(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 columnscolumn_name(stmt_id, idx) -> str: Column namecolumn_int(stmt_id, idx) -> int: 32-bit integer columncolumn_double(stmt_id, idx) -> float: 64-bit float columncolumn_text(stmt_id, idx) -> str: Text columncolumn_value(stmt_id, idx) -> str: Auto type conversion (int / float / text / null → str)column_type(stmt_id, idx) -> int: ReturnsSQLITE_INTEGER/SQLITE_FLOAT/SQLITE_TEXT/SQLITE_NULL/SQLITE_BLOBcolumn_is_null(stmt_id, idx) -> bool: Check whether the current column value is NULL
Metadata
libversion() -> str: SQLite C library versionchanges(conn_id) -> int: Number of rows affected by the most recent UPDATE/DELETESQLITE_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 formatplughw:CARD=X,DEV=Y, obtainable viagetInputDeviceList()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 stereochunk- Number of audio frames read each time; affects latency and memory usageformat- Audio format code, see table belowdevice_name- ALSA device name, obtained viagetInputDeviceList()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
callbackfirst parameter isbytes→setBytesCallback, 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 bytessample_rate- Sample rate (e.g. 16000)device_name- Device name, in the formatplughw:CARD=X,DEV=Y, obtainable viagetOutputDeviceList()
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 Hzchannels- Number of channelsformat- Audio format code, see table belowdevice_name- ALSA device name, obtained viagetOutputDeviceList()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
callbackreturn type isbytes→setBytesCallback, 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; recommendedhw:- 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 bytesfilename- File path to savesample_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(...) |
|
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