Python示例：使用sqlalchemy连接MemFire Cloud云数据库

[MemFireDB Robot]

# -*- coding: utf-8 -*-
"""
Copyright (c) 2024, Nimblex Co .,Ltd.

Created on 2024-05-06 15:42

备注：
1、python 3.8
2、在虚拟环境中使用pip安装psycopg2（macos用户请安装psycopg2-binary）、SQLAlchemy
"""
import threading
import sys

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, text
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.exc import OperationalError, SQLAlchemyError


Base = declarative_base()


class Counter(Base):
    __tablename__ = 'counters'

    id = Column(Integer, primary_key=True)
    counter = Column(Integer)


engine = None


def init_db_engine(host, port, dbname, user, password):
    global engine
    uri = f"postgresql://{user}:{password}@{host}:{port}/{dbname}"
    engine = create_engine(uri, echo=False,
                           pool_size=100, pool_recycle=3600,
                           pool_pre_ping=True)


def get_session():
    global engine
    SessionCls = sessionmaker(bind=engine)
    return SessionCls()


def test_transaction_try_again():
    def retryer():
        while 1:
            try:
                session = get_session()
                cnt = session.query(Counter).first()
                cnt.counter = cnt.counter + 1
                session.add(cnt)
                session.commit()
            except OperationalError as e:
                message = repr(e)
                if ('Try again' in message or '40001' in message or 'Restart read required' in message):
                    print('try again')
                    continue

            except SQLAlchemyError as e:
                message = repr(e)
                print("An error occurred: ", message)
                session.rollback()
                raise e

            finally:
                session.close()
            return
                

    threads = []
    for x in range(5):
        t = threading.Thread(target=retryer)
        t.start()
        threads.append(t)

    for t in threads:
        t.join()


def main():
    host = ""
    port = "5433"
    dbname = ""
    user = ""
    password = ""

    init_db_engine(host, port, dbname, user, password)
    Base.metadata.create_all(engine)

    session = get_session()
    try:
        # 删除并新增数据
        session.execute(text('delete from counters'))
        cnt = Counter(counter=999)
        session.add(cnt)
        session.commit()

        # 查询数据
        # res = session.execute(text('select * from counters;'))
        # rows = res.fetchall()
        # for row in rows:
        #     print(row)
    except SQLAlchemyError as e:
        print("An error occurred: ", e)
        session.rollback()
    finally:
        session.close()

    test_transaction_try_again()


if __name__ == '__main__':
    main()