Класс SingletonThreadPool

Класс SingletonThreadPool создает пул соединений, который поддерживает ровно одно соединение для каждого потока. При запросе соединения из потока пул проверяет, есть ли уже соединение, привязанное к этому потоку. Если соединение существует, оно возвращается повторно. Если соединения нет, создается новое и сохраняется в словаре по идентификатору потока. Такой пул удобен для SQLite при работе с файловой базой в многопоточном режиме, когда каждый поток должен использовать собственное соединение. Первым параметром передается функция создателя соединения creator, которая возвращает объект соединения с базой данных. Параметр pool_size задает максимальное количество соединений в пуле и по умолчанию равен 5.

Синтаксис

SingletonThreadPool(creator, [pool_size])

Пример

Давайте создадим пул SingletonThreadPool и получим из него соединение:

import sqlite3
from sqlalchemy.pool import SingletonThreadPool

def creator():
	return sqlite3.connect(':memory:')

pool = SingletonThreadPool(creator)
conn = pool.connect()
print(conn)
conn.close()

Результат выполнения кода:

"<sqlite3.Connection object at ...>"

Пример

Давайте убедимся, что в одном потоке пул возвращает одно и то же соединение:

import sqlite3
from sqlalchemy.pool import SingletonThreadPool

def creator():
	return sqlite3.connect(':memory:')

pool = SingletonThreadPool(creator)
conn1 = pool.connect()
conn2 = pool.connect()
print(conn1 is conn2)
conn1.close()
conn2.close()

Результат выполнения кода:

True

Пример

Давайте создадим движок SQLite, который использует SingletonThreadPool через параметр poolclass:

from sqlalchemy import create_engine
from sqlalchemy.pool import SingletonThreadPool

engine = create_engine(
	'sqlite:///:memory:',
	poolclass=SingletonThreadPool
)
print(engine.pool)

Результат выполнения кода:

"SingletonThreadPool(size=5)"

Пример

Давайте выполним запрос через движок с SingletonThreadPool:

from sqlalchemy import create_engine, text
from sqlalchemy.pool import SingletonThreadPool

engine = create_engine(
	'sqlite:///:memory:',
	poolclass=SingletonThreadPool
)
with engine.connect() as conn:
	res = conn.execute(text('SELECT 1'))
	print(res.scalar())

Результат выполнения кода:

1

Смотрите также

класс SingletonThreadPool,
который хранит по одному соединению на поток
класс StaticPool,
который хранит единственное соединение для всех потоков
класс QueuePool,
который поддерживает очередь соединений
функцию create_engine,
которая создает движок с указанным пулом