金仓数据库KingbaseES客户端编程开发框架-SQLAlchemy（4. 程序示例）

4. 程序示例

• 连接数据库

• ORM基本操作

4.1. 连接数据库

from sqlalchemy import create_engine

database = "TEST"
user = "SYSTEM"
password = "123456"
host = "127.0.0.1"
port = "54321"

conn_str = 'kingbase+ksycopg2://{}:{}@{}:{}/{}'.format(user, password, host, port, database)
engine = create_engine(conn_str)

conn = engine.connect()
res = conn.execute("select version()")
rows = res.fetchall()
print("database version:", rows)

4.2. ORM基本操作

import sqlalchemy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import sessionmaker

from sqlalchemy import create_engine

database = "TEST"
user = "SYSTEM"
password = "123456"
host = "127.0.0.1"
port = "54321"


def getengine():
    conn_str = 'kingbase+ksycopg2://{}:{}@{}:{}/{}'.format(user, password, host, port, database)
    engine = create_engine(conn_str)
    return engine


class test_sqlalchemy(declarative_base()):
    __tablename__ = 'test_sqlalchemy'
    num = Column(Integer, primary_key=True)
    val = Column(String)

    def __repr__(self):
        return "(Column: num={}, val={})".format(self.num, self.val)


if __name__ == "__main__":
    engine = getengine()
    classa1 = declarative_base()
    classa1.metadata.create_all(engine)
    session = sessionmaker(bind=engine)()

    # insert
    data0 = test_sqlalchemy(num='1', val='ABCDEF123')
    data1 = test_sqlalchemy(num='100', val='ORM方式插入数据')
    data2 = test_sqlalchemy(num='110', val='中文测试数据')

    session.add(data0)
    session.add(data1)
    session.add(data2)

    # update
    session.query(test_sqlalchemy).filter(test_sqlalchemy.num == 1).update(
        {"val": test_sqlalchemy.val + "(append data)"},
        synchronize_session=False
    )
    session.commit()

    # select
    results = session.query(test_sqlalchemy).all()
    print(results)

    session.close()