renom_cn.api.db

Copyright 2019, Grid.

This source code is licensed under the ReNom Subscription Agreement, version 1.0. ReNom Subscription Agreement Ver. 1.0 ( https://www.renom.jp/info/license/index.html )

class renom_cn.api.db. Connector

ベースクラス: object

Abstract class of db connector with sqlalchemy.

engine

Returns:

engine : sqlalchemy.engine.Engine

return attribute of sqlalchemy.engine.Engine instance.

get_all ( obj )

Get all object.

Args:

obj : sqlalchemy.ext.declarative.api.DeclarativeMeta class object.

Returns:

obj_list : list of sqlalchemy.ext.declarative.api.DeclarativeMeta instance.

insert ( obj )

Insert object.

Args:

obj : instance of sqlalchemy.ext.declarative.api.DeclarativeMeta class object.

Returns:

obj : instance of inserted sqlalchemy.ext.declarative.api.DeclarativeMeta class object.

session

Returns:

session : sqlalchemy.orm.session.Session

return attribute of sqlalchemy.orm.session.Session instance.

update ( obj )

Update object. Same with insert function.

Args:

obj : instance of sqlalchemy.ext.declarative.api.DeclarativeMeta class object.

Returns:

obj : instance of updated sqlalchemy.ext.declarative.api.DeclarativeMeta class object.

class renom_cn.api.db. SQLiteConnector ( host='' , database=':memory:' )

ベースクラス: renom_cn.api.db.Connector

Connector for SQLite.

Examples:

>>> from sqlalchemy import Column, Integer, TEXT
>>> from sqlalchemy.ext.declarative import declarative_base
>>> from renom_cn.api.db import SQLiteConnector
>>> Base = declarative_base()
>>> class Test(Base):
...     __tablename__ = 'test'
...     id = Column(Integer, primary_key=True, autoincrement=True)
...     name = Column(TEXT)
...
>>> conn = SQLiteConnector()
>>> Base.metadata.create_all(bind=conn.engine)
>>> data = Test(name="test1")
>>> conn.insert(data)
<__main__.Test object at 0x104bbbf28>
>>> conn.get_all(Test)  # get all Test table object.
[<__main__.Test object at 0x104bbbf28>]
>>> conn.session.query(Test).one()  # get one with session attribute.
<__main__.Test object at 0x104bbbf28>
>>> conn.session.query(Test).filter(Test.name=="test1").one()  # select with filter.
<__main__.Test object at 0x104bbbf28>

class renom_cn.api.db. MySQLConnector ( host , port , user , password , database )

ベースクラス: renom_cn.api.db.Connector

Connector for MySQL.

Examples:

>>> from sqlalchemy import Column, Integer, TEXT
>>> from sqlalchemy.ext.declarative import declarative_base
>>> from renom_cn.api.db import SQLiteConnector
>>> Base = declarative_base()
>>> class Test(Base):
...     __tablename__ = 'test'
...     id = Column(Integer, primary_key=True, autoincrement=True)
...     name = Column(TEXT)
...
>>> conn = MySQLConnector(host="localhost", port="3306",
...                       user="testuser", password="testpassword", database="testdb")
>>> Base.metadata.create_all(bind=conn.engine)
>>> data = Test(name="test1")
>>> conn.insert(data)
<__main__.Test object at 0x108a96550>
>>> conn.get_all(Test)  # get all Test table object.
[<__main__.Test object at 0x108a96550>]
>>> conn.session.query(Test).one()  # get one with session attribute.
<__main__.Test object at 0x108a96550>
>>> conn.session.query(Test).filter(Test.name=="test1").one()  # select with filter.
<__main__.Test object at 0x108a96550>

class renom_cn.api.db. PostgreSQLConnector ( host , port , user , password , database )

ベースクラス: renom_cn.api.db.Connector

Connector for PostgreSQL.

Examples:

>>> from sqlalchemy import Column, Integer, TEXT
>>> from sqlalchemy.ext.declarative import declarative_base
>>> from renom_cn.api.db import PostgreSQLConnector
>>> Base = declarative_base()
>>> class Test(Base):
...     __tablename__ = 'test'
...     id = Column(Integer, primary_key=True, autoincrement=True)
...     name = Column(TEXT)
...
>>> conn = PostgreSQLConnector(host="localhost", port="5432",
...                            user="testuser", password="testpassword", database="testdb")
>>> Base.metadata.create_all(bind=conn.engine)
>>> data = Test(name="test1")
>>> conn.insert(data)
<__main__.Test object at 0x10832e668>
>>> conn.get_all(Test)  # get all Test table object.
[<__main__.Test object at 0x10832e668>]
>>> conn.session.query(Test).one()  # get one with session attribute.
<__main__.Test object at 0x10832e668>
>>> conn.session.query(Test).filter(Test.name=="test1").one()  # select with filter.
<__main__.Test object at 0x10832e668>