Flask Web开发入门（六）之访问数据库

冷不防 2022-06-02 07:56 201阅读 0赞

我们介绍两种访问MySQL数据库的方式，一种是使用**mysql.connector**直连，另一种是使用**sqlalchemy** ORM框架。

*  mysql.connector直连方式

# get database connection
    def get_connection():
        try:
            conn = mysql.connector.connect(**config)
            return conn
        except Exception as e:
            logger.debug("Exception is %s" % e)
            return None
    
    
    # get user by user_id
    def get_user(user_id):
        try:
            conn = get_connection()
            if conn:
                cursor = conn.cursor()
                cursor.execute(user_sql, (user_id,))
                result = cursor.fetchall()
                logger.debug("conn is %s" % conn)
                return result
            else:
                logger.debug("conn is %s" % conn)
                return None
            logger.debug("conn is %s" % conn)
            return None
        except Exception as e:
            logger.debug("Exception is %s" % e)
            return None
        finally:
            cursor.close()
            conn.close()

其中，config定义如下：

> config = \{  
> ‘host’: ‘127.0.0.1’,  
> ‘user’: ‘user’,  
> ‘password’: ‘password’,  
> ‘port’: 3306,  
> ‘database’: ‘db’,  
> ‘charset’: ‘utf8’  
> \}  
>   
> user\_sql = ‘select \* from db.test where user\_id = %s limit 1’

*  **sqlalchemy** ORM框架

定义连接串

orm_url = 'mysql+mysqlconnector://user:password@ip:port/db'

获取用户对象

def get_user_engine(user_id):
        try:
            engine = create_engine(orm_url)
            session = sessionmaker()
            session.configure(bind=engine)
            Base.metadata.create_all(engine)
            s = session()
            ret = s.query(User).filter_by(username=user_id).first()
            return ret
        except Exception as e:
            logger.debug("Exception is %s" % e)
            return None

其中，用户类定义如下：

from sqlalchemy import Column, Integer, String
    from sqlalchemy.ext.declarative import declarative_base
    
    Base = declarative_base()
    
    
    # http://docs.sqlalchemy.org/en/latest/orm/mapping_columns.html
    class User(Base):
        __tablename__ = 't_user'
    
        id = Column('id', Integer, primary_key=True)
        username = Column('user_id', String(128))
        email = Column('email', String(128))
        password = Column('pwd_hash', String(128))
        create_time = Column('create_time', String(128))
        update_time = Column('update_time', String(128))
    
        def __init__(self, id, username, email, password, create_time, update_time):
            self.id = id
            self.username = username
            self.email = email
            self.password = password
            self.create_time = create_time
            self.update_time = update_time
    
        def __repr__(self):
            return '<id is %s, username is %s, password is %s, email is %s, create time is %s, update time is %s>' % (
                self.id, self.username, self.password, self.email, self.create_time, self.update_time)

调用方法：

user = flask_db.get_user_engine(username)
             logger.debug('db user id is %s, detail is %s' % (user.username, user))

[http://docs.sqlalchemy.org/en/latest/orm/mapping\_columns.html][http_docs.sqlalchemy.org_en_latest_orm_mapping_columns.html]

**源码参考：**[https://github.com/ypmc/flask-sqlalchemy-web][https_github.com_ypmc_flask-sqlalchemy-web]

[http_docs.sqlalchemy.org_en_latest_orm_mapping_columns.html]: http://docs.sqlalchemy.org/en/latest/orm/mapping_columns.html
[https_github.com_ypmc_flask-sqlalchemy-web]: https://github.com/ypmc/flask-sqlalchemy-web