Run ray multi threading with sqlalchemy and sqlite database, in python.

main.py

from sqlalchemy import create_engine

from sqlalchemy import select, insert, func

from tables import Base, Record

from time import time

import ray
import uuid


@ray.remote
class Worker:
    def __init__(self, i: int) -> None:
        self.i = i

    def do_write(self, x: str):
        # print(self.i, "inserting", x)

        engine = create_engine("sqlite:///database.db")
        conn = engine.connect()
        stmt = insert(Record)

        conn.execute(stmt, [{"value": x}])
        conn.commit()

        return x


if __name__ == "__main__":
    ray.init()

    engine = create_engine("sqlite:///database.db")

    Base.metadata.create_all(engine)

    data = [str(uuid.uuid4()) for _ in range(10000)]

    workers = [Worker.remote(w) for w in range(32)]
    actor_pool = ray.util.ActorPool(workers)

    results = actor_pool.map(lambda a, x: a.do_write.remote(x), data)  # type: ignore

    start_time = time()

    i = 0
    for r in results:
        if i % 1000 == 0:
            interval = time() - start_time
            print(i, f"{interval:.2f} (s)", f"{i/interval:.2f} (it/s)")
        i += 1

    interval = time() - start_time

    print(interval, "s")

    with engine.connect() as conn:
        res = conn.execute(select(func.count()).select_from(Record))

        print(res.scalar())

tables.py

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy import String, DateTime
from sqlalchemy.orm import mapped_column, Mapped
from sqlalchemy.sql.functions import now

from datetime import datetime


class Base(DeclarativeBase):
    pass


class Record(Base):
    __tablename__ = "records"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    creation_time: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=now())
    value: Mapped[str] = mapped_column(String)