zoukankan      html  css  js  c++  java
  • python 操作mysql数据库

    安装mysql模块

    pip install PyMySQL
    pip install sqlalchemy

    操作mysql数据 

    #!/usr/bin/env python
    # -*- coding: utf-8 -*-
    # import cx_Oracle
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    import traceback
    
    class DB:
        """
        数据库操作
        create_engine("dialect+driver://username:password@host:port/database") #初始化连接
        dialect:数据库类型,包括sqlite,mysql,postgresql,oracle,mssql等
        driver:指定连接数据库的API,如 psycopg2,pyodbc,cx_oracle,为可选关键字
        """
        engine = None
            
        def session(self):
            try:
                dbSession = sessionmaker(bind=self.engine)
                cursor = dbSession()
            except Exception as e :
                print('DataBase found errors: {0}'.format(e))
            return cursor    
        
        def query(self,sql):
            #执行查询语句
            try:
                query = self.session()
                ret = query.execute(sql)
                data = ret.fetchall() 
                return data
            except:
                print(traceback.format_exc())
                print('查询数据异常')
                return []
            
        def execute(self,sql):
            #执行单条语句
            try:
                execute = self.session()
                ret = execute.execute(sql)
                execute.commit()
                execute.close()            
                return 0
            except:
                print('执行[%s]失败'%sql)
                execute.rollback()
                execute.close()
                return -1 
            
        def execute_many(self,SqlList):
            #执行多条sql
            try:
                execute = self.session()
                for sql in SqlList:
                        ret = execute.execute(sql)
                execute.commit()
                execute.close()
                return 0
            except:
                print('执行[%s]失败'%sql)
                execute.rollback()
                execute.close()            
                return -1 
            
    class MysqlDB(DB):
        """mysql数据库"""
        def __init__(self,host,user,passwd,database,port=3306):
            url = f'mysql+pymysql://{username}:{passwd}@{host}:{port}/{database}?charset=utf8'
    self.engine = create_engine(url,encoding='utf8',pool_size=100, pool_recycle=3600, echo=False) class OracleDB(DB): """Oracle数据库""" def __init__(self,host,username,passwd,dsn,port=1521): url = f'oracle+cx_oracle://{username}:{passwd}@{host}:{port}/{dsn}?charset=utf8'
    self.engine = create_engine(url,encoding='utf8',pool_size=100, pool_recycle=3600, echo=False) if __name__ == '__main__': mysql = MysqlDB('39.101.187.225','boye','boyecom@126','crawler') ret = mysql.query('show tables') query =mysql.query("select * from test"); print(ret) print(query)
  • 相关阅读:
    python 获取文件大小,创建时间和访问时间
    url中的20%、22%、26%、7B%、%7D、28%、29% 代表真实的字符(转发)
    python3 采集需要登录的网页数据
    python3 模拟鼠标中轴滚动
    python3 爬虫小技巧,
    马拉松中级训练计划
    python 使用夜神模拟器
    Python3+mitmproxy安装使用教程(Windows)(转载)
    adb server version (31) doesn't match this client (41); killing...
    appium+python环境搭建
  • 原文地址:https://www.cnblogs.com/boye169/p/13904468.html
Copyright © 2011-2022 走看看