thread

#!/usr/bin/env python3

import sys
import traceback

from concurrent.futures import ThreadPoolExecutor
from operator import itemgetter
from threading import get_ident

import orjson


DATA = sorted(
    [
        {
            "id": i,
            "name": "90as90ji0123ioj2390as90as90",
            "body": "哈哈89asu89as😊89as9as90jas-😋0apjzxiojzx89hq23n",
            "score": 901290129.1,
            "bool": True,
            "int": 9832,
            "none": None,
        }
        for i in range(10)
    ],
    key=itemgetter("id"),
)


STATUS = 0

TEST_MESSAGE = "thread test running..."

sys.stdout.write(TEST_MESSAGE)
sys.stdout.flush()


def test_func(n):
    try:
        assert sorted(orjson.loads(orjson.dumps(DATA)), key=itemgetter("id")) == DATA
    except Exception:
        STATUS = 1
        traceback.print_exc()
        print("thread %s: %s dumps, loads ERROR" % (get_ident(), n))


with ThreadPoolExecutor(max_workers=4) as executor:
    executor.map(test_func, range(200000), chunksize=1000)
    executor.shutdown(wait=True)


if STATUS == 0:
    sys.stdout.write(f"\r{TEST_MESSAGE} ok\n")
else:
    sys.stdout.write(f"\r{TEST_MESSAGE} error\n")


sys.exit(STATUS)