Flask 에서 SQLAlchemy 사용시 Query 팁
설정 Connection 기본 DBMS 연결 정보
# MySQL app.config['SQLALCHEMY_DATABASE_URI'] = r'mysql+pymysql://user:passwd@address:3306/db_name?charset=UTF8MB4' db = SQLAlchemy() db.init_app(app) Bind 추가 기본 DBMS 이외, 추가적인 DB 정보 추가
app.config['SQLALCHEMY_BINDS'] = { 'dbms1': r'mysql+pymysql://user:passwd@address1:3306/db_name?charset=UTF8MB4', 'dbms2': r'mysql+pymysql://user:passwd@address2:3306/db_name?charset=UTF8MB4' } # Model Example class Network(db.Model): __tablename__ = 'network' __bind_key__ = 'dbms1' ip = db.Column(db.String(128), primary_key=True) switch = db.Column(db.String(128)) doc = db.Column(db.JSON) AlchemyEncoder SQLAlchemy JSON Encoder
class AlchemyEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj.__class__, DeclarativeMeta): # an SQLAlchemy class fields = {} for field in [x for x in dir(obj) if not x.startswith('_') and x != 'metadata']: data = obj.__getattribute__(field) try: json.dumps(data) # this will fail on non-encodable values, like other classes fields[field] = data except TypeError: if isinstance(data, datetime): fields[field] = data.strftime('%Y-%m-%d %H:%M:%S') elif isinstance(data, date): fields[field] = data.strftime('%Y-%m-%d') else: fields[field] = None return fields return json.JSONEncoder.default(self, obj) Flask Basic Route @app.route('/api/network/') def api_network(): cur = db.session.query(Network) return json.dumps({ 'data': cur.all() }, cls=AlchemyEncoder) DB Model Generate # 해당 DB 전체 Table 대상 $ ./venv/bin/flask-sqlacodegen 'mysql+pymysql://user:passwd@address2:3306/db_name' --flask # 해당 DB 테이블 지정 $ ./venv/bin/flask-sqlacodegen 'mysql+pymysql://user:passwd@address2:3306/db_name' --flask --table network,other_table Query Sample Select All cur = db.session.query(Network) Select Filter, Sort # Filter cur = db.session.query(network).filter(network.switch == switch_name) cur = db.session.query(network).filter(network.ip.like('{}%'.format(ip))) cur = db.session.query(network).filter(network.switch.in_(('AA', 'BB'))) # Text filter, json cur = db.session.query(network).filter(text(r'''doc->>"$.name" <> '' ''')) # OR cur = db.session.query(network).filter(or_(network.ip == ip, network.switch == switch_name)) # AND cur = db.session.query(network).filter(and_(network.ip == ip, network.switch == switch_name)) # SORT cur = db.session.query(network).filter(network.switch == switch_name) \ .order_by(network.ip.desc()) Join # Join cur = db.session.query(Network, NetworkSwitch) \ .filter(Network.ip == NetworkSwitch.ip, Network.switch == NetworkSwitch.switch) # Outer Join cur = db.session.query(Network) \ .outerjoin(NetworkSwitch, and_(Network.ip == NetworkSwitch.ip, Network.switch == NetworkSwitch.switch)) # Self Join, Table Alias NetworkAlias = aliased(Network) cur = db.session.query(Network, NetworkAlias) \ .filter(Network.ip == NetworkAlias.ip, Network.switch == NetworkAlias.switch)