Navigation

    MemFireDB论坛

    • Register
    • Login
    • Search
    • Categories
    • Recent
    • Tags
    • Popular
    • Users
    • Groups

    python使用SQLAlchemy访问MemFireDB

    MemFireDB新手区
    1
    1
    26
    Loading More Posts
    • Oldest to Newest
    • Newest to Oldest
    • Most Votes
    Reply
    • Reply as topic
    Log in to reply
    This topic has been deleted. Only users with topic management privileges can see it.
    • M
      MemFireDB Robot last edited by

      SQLAlchemy使用证书连接MemFireDB的例子,创建完数据库后,把下载的证书解压到代码同级的目录就可以了,记得将host、port、dbname、dbuser设置为对应的值:

      # -*- coding: utf-8 -*-
      
      import sys
      
      from sqlalchemy.ext.declarative import declarative_base
      from sqlalchemy import Column, Integer
      from sqlalchemy import create_engine
      from sqlalchemy.orm import sessionmaker
      import sqlalchemy.engine.url as url
      
      
      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, ssl_ca, ssl_cert, ssl_key):
          global engine
          uri = url.URL(
              drivername="postgresql",
              host=host,
              port=port,
              username=user,
              database=dbname,
          )
      
          ssl_args = {
              "sslmode": "verify-ca",
              "sslrootcert": ssl_ca,
              "sslcert": ssl_cert,
              "sslkey": ssl_key
          }
          print ssl_args
          engine = create_engine(uri, connect_args=ssl_args, encoding='utf-8', echo=False,
                                 pool_size=100, pool_recycle=3600, pool_pre_ping=True)
      
      
      def get_session():
          global engine
          SessionCls = sessionmaker(bind=engine)
          return SessionCls()
      
      
      def main():
          host = sys.argv[1]
          port = int(sys.argv[2])
          dbname = sys.argv[3]
          dbuser = sys.argv[4]
      
          init_db_engine(host, port, dbname, dbuser, './root.crt', './memfiredb.crt', './memfiredb.key')
          Base.metadata.create_all(engine)
      
          session = get_session()
          session.execute('delete from counters')
          cnt = Counter(counter=1)
          session.add(cnt)
          session.commit()
          session.close()
      
      
      if __name__ == '__main__':
          main()
      
      
      1 Reply Last reply Reply Quote 2
      • First post
        Last post