# -*- coding:UTF -8-*-import

osimport

pandas as pd#

检查是否有pymysql这个包存在import

pymysql#

SQLAlchemy本身无法操作数据库，其必须以来pymsql等第三方插件。from

sqlalchemy import create_engine,Sequence,textfrom

sqlalchemy.ext.declarative import declarative_basefrom

sqlalchemy import Column,

Integer,

String,

ForeignKey,

UniqueConstraint,

Indexfrom

sqlalchemy.orm import sessionmaker,

relationship#

设置工作路径os.chdir('E:

/data')#

读取文件sale

= pd.read_csv('sale.csv',encoding

= 'GB18030')'''

sale.csv:

year

market sale

profit

2010

东33912

2641

2010

南32246

2699

2010

西34792

2574

2010

北31884

2673

2011

东31651

2437

2011

南30572

2853

2011

西34175

2877

2011

北30555

2749

2012

东31619

2106

2012

南32443

3124

2012

西32103

2593

2012

北31744

2962

'''#

初始化数据连接数据库的信息host

= yourhostuser

= youridpassword

= yourpassworddb

= yourdatbasenameport

= 3306charset

= 'utf8'#

连接数据库#

pool_recycle控制任何连接的最大连接时长(秒)；pool_size连接池的最大连接数，默认5，设置0时为无限制连接数连接；#

max_overflow溢出连接池的最大额外连接数，超过pool_size所设置的连接数的溢出连接数。实际上总的并行连接总数为pool_size

# +max_overflow,设置-1为不设置溢出最大限制，默认为10；pool_timeout等待返回信息时间，默认为30(秒)#

echo类似于python的logging(日志)功能，默认为False

# 'mysql+pymysql'，mysql为数据库类型，pymysql为连接方式engine

= create_engine('mysql+pymysql://{user}:{password}@{host}:{port}/{db}?charset={charset}'

.format(user

= user,

host = host,

password = password,

db = db,port

= port,

charset = charset),

pool_size = 30,max_overflow

= 0)#

创建直接调用数据库属性conn

= engine.connect()#

查询数据库版本号#

execute里面为原始sql语句；#

fetchone为返回第一行结果；fetchall()为返回所有结果;fetchmany(n)为返回第n行结果databaseVersion

= conn.execute("SELECT

VERSION()").fetchone() # ##################使用原始sql语句运行#####################

# 创建一个数据表conn.execute('CREATE

TABLE sale ('

'year CHAR(20)

NOT NULL,'

'market CHAR(20),'

'sale INT ,'

'profit_from_sale INT )')#

删除表conn.execute('DROP

TABLE sale')#

将df存进数据表中且如果有同名数据表就把之前的删除后重新添加sale.to_sql('sale',conn,if_exists='replace',index

= False)#

查询数据表语句conn.execute('select

* from sale').fetchall()

conn.execute('select

DISTINCT year,market from sale').fetchall()

conn.execute("select

* from sale where year=2012 and market=%s",'东').fetchall()

conn.execute('select

year,market,sale,profit from sale order by sale').fetchall()#

添加数据记录conn.execute("insert

into sale (year,market,sale) value (2010,'东',33912)")#

删除数据记录conn.execute("delete

from sale") # #########################ORM模式############################

##创建表#

创建对象的基类Base

= declarative_base()#

在本地内存创建单表(建立本地与sql的连接)class

Sale(Base):

__tablename__ = 'sale'year

= Column(String(20),

primary_key=True,

nullable=False)

market = Column(String(10))

sale = Column(Integer)

profit_from_sale = Column(Integer)

# 测试用

def __repr__(self):

return '%s(%s,%s,%s,%s)'%\

(self.__tablename__,self.year,self.market,self.sale,self.profit_from_sale)

# 创建属性，UniqueConstraint为unique

key，Index为index

# __table_args__ = (

#

UniqueConstraint('id', 'name', name='uix_id_name'),

#

Index('ix_id_name', 'name', 'extra'),)'''

sale表在sql的DDL

CREATE TABLE `sale` (

`year`

varchar(20) NOT NULL,

`market`

varchar(10) DEFAULT NULL,

`sale`

int(11) DEFAULT NULL,

`profit_from_sale` int(11) DEFAULT NULL,

PRIMARY

KEY (`year`)

) ENGINE=InnoDB DEFAULT CHARSET=utf8;

'''#

查询本地表的结构Sale.__table__ # 创建一对多表(建立本地与sql的连接)class

Favor(Base):

__tablename__ = 'favor'

nid = Column(Integer,

primary_key=True,

autoincrement=True)

caption = Column(String(50),

default='red',

unique=True)

sale_year = Column(String(10),

ForeignKey("sale.year"))'''

favor表在sql的DDL

CREATE TABLE `favor` (

`nid`

int(11) NOT NULL AUTO_INCREMENT,

`caption`

varchar(50) DEFAULT NULL,

`sale_year` varchar(10) DEFAULT NULL,

PRIMARY

KEY (`nid`),

UNIQUE

KEY `caption` (`caption`),

KEY

`sale_year` (`sale_year`),

CONSTRAINT `favor_ibfk_1` FOREIGN KEY (`sale_year`) REFERENCES

`sale` (`year`)

) ENGINE=InnoDB DEFAULT CHARSET=utf8;

'''#

初始化数据库函数(全部)Base.metadata.create_all(conn)#

删除数据库(全部)Base.metadata.drop_all(conn) ##操作表#

创建session********************Session

= sessionmaker(bind=conn)

session = Session()#

新增单条数据new

= Sale(year=2010,

market='东',

sale=33912)

session.add(new)#

新增多条数据session.add_all([

Sale(year=2011,

market='东',

sale=31651,

profit_from_sale=2437),

Sale(year=2012,

market='东',

sale=31619),

])try:

# 提交即保存到数据库

session.commit()except:

# 回滚，返回session执行之前的状态

session.rollback()#

查询表(Querying)

# 返回的是list格式session.query(Sale.year,

Sale.sale).all()

# order_by以什么为顺序(默认升序)for

i in session.query(Sale).order_by(Sale.sale):

print(i.year,i.sale)

#first()为第一行，all()为全部for

i in session.query(Sale,Sale.year).all():

print(i.Sale,i.year)#

测试用

# label为设置替换名for

i in session.query(Sale.year.label('test')).all():

print(i.test)from

sqlalchemy.orm import aliased#

替换功能Sale_test

= aliased(Sale,

name='sale_test'

)#

name为table的名字for

i in session.query(Sale_test).all()[1:3]:

print(i)

# filter_by为匹配;filter也一样，用法为filter(Sale.year

= 2012),

# filter与filter之间用.连接，代表and的意思for

i in session.query(Sale).filter_by(year

= 2012).all():

print(i) # 使用text来混合原始sql语句使用session.query(Sale).filter(text('year>2010')).order_by(text('sale')).all()

session.query(Sale).filter(text('year=:year

and market=:market'))\

.params(year

= 2011,market

='东').order_by(Sale.sale).one()

session.query(Sale).from_statement(text('select

* from sale where year={}'.format(2012))).all() # Counting统计session.query(Sale.profit_from_sale).count()

# 统计行数,返回intfrom

sqlalchemy import func#

func 后可以跟任意函数名，只要该数据库支持session.query(func.count(Sale.profit_from_sale)).scalar()#

统计非nan值个数,返回intsession.query(func.count()).select_from(Sale).scalar()

session.query(Sale.profit_from_sale).distinct().count()#

统计不同的个数#

删除数据session.query(Sale).filter(Sale.year

== 2012).delete()

session.commit()#

修改数据(如果之前删除了数据，由于数据表发生变化，需要重新连接sql)session.query(Sale).filter(Sale.year

== 2010).update({"profit_from_sale":666})

session.query(Sale).filter(Sale.year == 2010).update({Sale.profit_from_sale:Sale.profit_from_sale+333})

session.commit()#

关闭session****************session.close() # ###################pandas模式###################

# 将DataFrame/Series写入sql里面#

sale1为写入表的名称;conn为SQLAlchemy/DBAPI2连接sql的引擎;schema为设定模式默认为默认模式；#

if_exists:{'fail','replace','append'}为如果存在相同的表，fail不操作不添加，#

replace删除原表重新添加，append插入数据(列名要一致,尾端插入)。#

三者在没有原表时都会创建新表，default

‘fail’；#

index：default

True，将DataFrame的index作为一列写入sql里；#

index_label：{string

or sequence}, default None，设置sql的索引所在的列(名)；#

chunksize:{int}, default None,设置每次写入sql里的DF行数，若为空，则一次心写入全部；#

dtype:{dict of column name to SQL type}, default

None,写入sql里面的列解释，必须是在#

SQLAlchemy模式下sale.to_sql('sale1',conn,if_exists='append',index

= 0,index_label='year',chunksize

= None) # 将sql里面的数据读取到DF

# schema为sql的模式;index_col为指定sql里的一列为DF的index；#

coerce_float将非数字非字符串自动转化为浮点数，默认为True；#

parse_dates解析列属性，输入List

of column names将对应的列内容转化为时间日期格式。#

输入Dict{column_name:

format string}将列内容转化为对应的字符串格式.

# 输入Dict{column_name:

arg dict}将列内容转化为pandas.to_datetime()格式,

# arg dict为pandas.to_datetime()参数;

# columns为需要查询sql的table的列内容所对应的列名，默认None为全查readSqlTable1

= pd.read_sql_table('sale1',conn,parse_dates=('year'))type(readSqlTable1['year'][0])#pandas._libs.tslib.TimestampreadSqlTable2

= pd.read_sql_table('sale1',conn,parse_dates={'year':'%Y-%M-%D'},columns=['year','sale'])type(readSqlTable2['year'][0])#datetime.datereadSqlTable3

= pd.read_sql_table('sale1',conn,parse_dates={'year':{'format':'%Y-%M-%D'}})type(readSqlTable3['year'][0])#datetime.date

# 参数同上readSqlQuery

= pd.read_sql_query('select

* from sale1',conn)

readSql1 = pd.read_sql('select

* from sale1',conn)

readSql2 = pd.read_sql('sale1',conn) # 关闭sql连接conn.close()