文档章节

Python连接mysql或postgres(sqlalchemy)

wangxuwei
 wangxuwei
发布于 2016/04/23 21:26
字数 1879
阅读 946
收藏 2

安装

sudo pip install sqlalchemy 或sudo pip3 install sqlalchemy

下载速度慢可使用国内源,如:

sudo pip install sqlalchemy  -i http://mirrors.aliyun.com/pypi/simple --trusted-host mirrors.aliyun.com

测试1

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

eng = create_engine('mysql+mysqlconnector://y-user:y-passwd@127.0.0.1:3306/y-db?charset=utf8')
#eng = create_engine('postgresql+psycopg2://y-user:y-passwd@127.0.0.1:5432/y-db',echo=True,client_encoding='utf8')

DB_Session = sessionmaker(bind=eng)
session = DB_Session()
data = session.execute("SELECT * FROM y-table")
for row in data:  
    for col in row:  
       print col,  
    print  
session.close()

 

测试2

#coding=utf-8
from sqlalchemy import *
from sqlalchemy.orm import *

eng = create_engine('mysql+mysqlconnector://y-user:y-passwd@127.0.0.1:3306/y-db?charset=utf8')
metadata = MetaData(eng) # 绑定元信息
y_table = Table('y-table', metadata, autoload=True)
session = create_session()
query = session.query(y_table)
data = query.all()
for row in data:
    for col in row:
       print col,
    print
session.close()

 

测试3--orm方式

数据库表STUDENT的建立参考:

http://my.oschina.net/u/2245781/blog/653897?fromerr=YEVmVLAx

# coding=utf-8
from sqlalchemy import Column, String, create_engine
from sqlalchemy.dialects.mysql import DATETIME
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
import sys
reload(sys)
sys.setdefaultencoding( "utf-8" )
# 创建对象的基类:
Base = declarative_base()
# 定义User对象:
class Stu(Base):
    # 表的名字:
    __tablename__ = 'STUDENT'

    # 表的结构:
    SNO = Column(String(7), primary_key=True)
    SNAME = Column(String(8))
    SEX = Column(String(2))
    BDATE = Column(DATETIME)
    DIR = Column(String(16))
    
    def __repr__(self):
        return "<Stu(SNO='%s', SNAME='%s', SEX='%s',BDATE ='%s',DIR ='%s')>" % \
           (self.SNO,self.SNAME,self.SEX,self.BDATE.strftime("%Y-%m-%d"),self.DIR)
# 初始化数据库连接:
engine = create_engine('mysql+mysqlconnector://y-user:y-passwd@127.0.0.1:3306/y-db?charset=utf8')
# 创建DBSession类型:
DBSession = sessionmaker(bind=engine)
# 创建session对象:
session = DBSession()
query = session.query(Stu)
for s in session.query(Stu):
    print(s)
#for s in query.all():
#     print s.SNAME+' '+s.SNO+' '+s.SEX +' '+s.BDATE.strftime("%Y-%m-%d")+' ' + s.DIR
session.close()

插入

# coding=utf-8
from sqlalchemy import Column, String, create_engine
from sqlalchemy.dialects.mysql import DATETIME
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

# 创建对象的基类:
Base = declarative_base()
# 定义User对象:
class Stu(Base):
    # 表的名字:
    __tablename__ = 'STUDENT'

    # 表的结构:
    SNO = Column(String(7), primary_key=True)
    SNAME = Column(String(8))
    SEX = Column(String(2))
    BDATE = Column(DATETIME)
    DIR = Column(String(16))
    def __repr__(self):
        return "<Stu(SNO='%s', SNAME='%s', SEX='%s',BDATE ='%s',DIR ='%s')>" % \
           (self.SNO,self.SNAME,self.SEX,self.BDATE.strftime("%Y-%m-%d"),self.DIR)
    

# 初始化数据库连接:
engine = create_engine('mysql+mysqlconnector://y-user:y-passwd@127.0.0.1:3306/y-db?charset=utf8')
# 创建DBSession类型:
DBSession = sessionmaker(bind=engine)
# 创建session对象:
session = DBSession()
new_stu = Stu(SNO='9302205', SNAME='方邪真',SEX='男',BDATE='1973-11-18',DIR='数理逻辑')
# delete from STUDENT where SNO ='9302205';
# 添加到session:
session.add(new_stu)
# 提交即保存到数据库:
session.commit()
# 关闭session:
session.close()

根据sqlalchemy文档http://docs.sqlalchemy.org/en/latest/dialects/postgresql.html

当数据库为postgresql时DATETIME改为DATE类型,关于查询的例子如下:

# coding=utf-8
from sqlalchemy import Column, String, create_engine
from sqlalchemy.dialects.postgres import DATE
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from datetime import *
import sys
reload(sys)
sys.setdefaultencoding("utf-8")

# 创建对象的基类:
Base = declarative_base()
# 定义Stu对象:
class Stu(Base):
    # 表的名字:
    __tablename__ = 'student'

    # 表的结构:
    sno = Column(String(7), primary_key=True)
    sname = Column(String(8))
    sex = Column(String(2))
    bdate = Column(DATE)
    dir = Column(String(16))
    def __repr__(self):
        return "<Stu(SNO='%s', SNAME='%s', SEX='%s',BDATE ='%s',DIR ='%s')>" % \
               (self.sno,self.sname,self.sex,self.bdate.strftime("%Y-%m-%d"),self.dir)
# 初始化数据库连接:
engine = create_engine('postgresql+psycopg2://user:passwd@127.0.0.1:5432/db',echo=True,client_encoding='utf8')
# 创建DBSession类型:
DBSession = sessionmaker(bind=engine)
# 创建session对象:
session = DBSession()
query = session.query(Stu)
#for s in query.all():
#    print s.sname+' '+s.sno+' '+s.sex +' '+s.bdate.strftime("%Y-%m-%d")+' ' + s.dir
for s in session.query(Stu):
    print(s)
session.close()

根据官方文档中的例子:

表创建crtusers.py

# coding=utf-8
from sqlalchemy import ForeignKey
from sqlalchemy import create_engine
from sqlalchemy import Column, Integer,String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship

engine = create_engine('mysql+mysqlconnector://y-user:y-passwd@127.0.0.1:3306/y-db?charset=utf8')
Base = declarative_base()
#建立表users
class User(Base):
     __tablename__ = 'users'

     id = Column(Integer, primary_key=True)
     name = Column(String(20))
     fullname = Column(String(40))
     password = Column(String(12))

     def __init__(self, name, fullname, password):
         self.name = name
         self.fullname = fullname
         self.password = password

     def __repr__(self):
        return "<User(name='%s', fullname='%s', password='%s')>" % (self.name, self.fullname, self.password)
Base.metadata.create_all(engine)

#建立表addresses
class Address(Base):
   __tablename__ = 'addresses'
   id = Column(Integer, primary_key=True)
   email_address = Column(String(40), nullable=False)
   user_id = Column(Integer, ForeignKey('users.id'))

   user = relationship("User", back_populates="addresses")
   def __repr__(self):
      return "<Address(email_address='%s')>" % self.email_address

User.addresses = relationship("Address", order_by=Address.id, back_populates="user")
Base.metadata.create_all(engine)

#插入一条user记录
DBSession = sessionmaker(bind=engine)
session = DBSession()
ed_user = User(name='ed', fullname='Ed Jones', password='edspassword')
session.add(ed_user)
#插入多条记录
session.add_all([User(name='wendy', fullname='Wendy Williams', password='foobar'),\
                 User(name='mary', fullname='Mary Contrary', password='xxg527'),\
                 User(name='lisa', fullname='lisa Contrary', password='ls123'),\
                 User(name='cred', fullname='cred Flinstone', password='bla123'),\
                 User(name='fred', fullname='Fred Flinstone', password='blah')])
#插入1条user记录,两条address
jack = User(name='jack', fullname='Jack Bean', password='gjffdd')
jack.addresses = [Address(email_address='jack@google.com'),Address(email_address='j25@yahoo.com')]
session.add(jack)

# 提交即保存到数据库:
session.commit()
# 关闭session:
session.close()

查询

# coding=utf-8
from operator import or_
from sqlalchemy import text
from sqlalchemy import create_engine
from sqlalchemy import Column, Integer,String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import func

from crtusers import Address, User

engine = create_engine('mysql+mysqlconnector://y-user:y-passwd@127.0.0.1:3306/y-db?charset=utf8')
DBSession = sessionmaker(bind=engine)
session = DBSession()

#query sample
query = session.query(User)
for u in query.all():
    print str(u.id)+' '+u.name+' '+u.fullname+' '+u.password
for u in session.query(User).order_by(User.id)[1:4]:
    print(u)
for name in session.query(User.name).filter_by(fullname='Ed Jones'):
    print(name)
for u in session.query(User).filter(User.name.isnot(None)):
    print(u)
for u in session.query(User).filter(or_(User.name == 'ed', User.name == 'wendy')):
    print(u)
for u in session.query(User).filter(~User.name.like('%ed')).order_by(User.id):
    print(u)
for u in session.query(User).filter(User.name=='ed').filter(User.fullname == 'Ed Jones'):
    print(u)
for u in session.query(User).filter(User.name.in_(['ed', 'wendy', 'jack'])):
    print(u)

#Using Textual SQL
for user in session.query(User).filter(text("id<224")).order_by(text("id")).all():
    print(user.name)
for u in session.query(User).from_statement\
            (text("SELECT * FROM users where id >= :id")).\
            params(id=2).all():
        print(u)

for u in session.query(User).from_statement \
            (text("SELECT * FROM users where name=:name")). \
            params(name='ed').all():
        print(u)
#Counting on count()
print(session.query(func.count('*')).select_from(User).scalar())
print(session.query(User).filter(User.name.like('%ed')).count())
for n in session.query(func.count(User.name), User.name).group_by(User.name).all():
    print(n)
    
#Querying Address
for u, a in session.query(User, Address).filter(User.id == Address.user_id).\
        filter(Address.email_address == 'jack@google.com').all():
    print(u)
    print(a)

for u in session.query(User).join(Address).filter(Address.email_address=='jack@google.com').all():
    print(u)
for name, in session.query(User.name).filter(User.addresses.any(Address.email_address.like('%google%'))):
    print(name)
# 关闭session:
session.close()

安装psycopg2要指定路径,故用源码编译方式

psycopg2-2.6.2$ python setup.py build_ext --pg-config /opt/PostgreSQL/9.5/bin/pg_config build

psycopg2-2.6.2$ sudo python setup.py build_ext --pg-config /opt/PostgreSQL/9.5/bin/pg_config install

新版地址 

wget http://initd.org/psycopg/tarballs/PSYCOPG-2-7/psycopg2-2.7.1.tar.gz

例子:根据psycopg2-2.6.2/examples/simple.py(只修改DSN以适合你的数据库环境)

# simple.py - very simple example of plain DBAPI-2.0 usage
#
# currently used as test-me-stress-me script for psycopg 2.0
#
# Copyright (C) 2001-2010 Federico Di Gregorio  <fog@debian.org>
#
# psycopg2 is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# psycopg2 is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public
# License for more details.

## put in DSN your DSN string

DSN = 'dbname=your-db user=your-user password=your-passwd host=127.0.0.1'

## don't modify anything below this line (except for experimenting)

class SimpleQuoter(object):
    def sqlquote(x=None):
        return "'bar'"

import sys
import psycopg2

if len(sys.argv) > 1:
    DSN = sys.argv[1]

print "Opening connection using dsn:", DSN
conn = psycopg2.connect(DSN)
print "Encoding for this connection is", conn.encoding

curs = conn.cursor()
curs.execute("SELECT 1 AS foo")
print curs.fetchone()
curs.execute("SELECT 1 AS foo")
print curs.fetchmany()
curs.execute("SELECT 1 AS foo")
print curs.fetchall()

conn.rollback()

sys.exit(0)

curs.execute("SELECT 1 AS foo", async=1)

curs.execute("SELECT %(foo)s AS foo", {'foo':'bar'})
curs.execute("SELECT %(foo)s AS foo", {'foo':None})
curs.execute("SELECT %(foo)f AS foo", {'foo':42})
curs.execute("SELECT %(foo)s AS foo", {'foo':SimpleQuoter()})

PyGreSQL安装

下载:PyGreSQL-5.0.4.tar.gz

修改setup.py

    f = os.popen('pg_config --%s' % s)

改为:

    f = os.popen('/opt/PostgreSQL/10/bin/pg_config --%s' % s)

 再安装:

 sudo python3 setup.py install

测试代码PyGreSQLtest1.py:

#coding=utf-8
from pg import DB

db = DB(dbname='testdb', host='localhost', port=5432,user='mymotif', passwd='wxwpxh');
print(db.query("select * from userinfo"));

测试2:

# -*- coding: utf-8 -*-

#导入pg模块
import pg

def operate_postgre_tbl_product():
  #连接数据库 
  try:
    pgdb_conn = pg.connect(dbname = "testdb", host = "127.0.0.1", user = "mymotif", passwd = "wxwpxh")
  except Exception, e:
     print e.args[0]
     return  
   
   
  #删除表
  sql_desc = "DROP TABLE IF EXISTS tbl_product3;"
  try:
    pgdb_conn.query(sql_desc)
  except Exception, e:
    print "drop table failed"
    pgdb_conn.close() 
    return
  
  #创建表
  sql_desc = """CREATE TABLE tbl_product3(
    i_index INTEGER,
    sv_productname VARCHAR(32)
    );"""
  try:  
    pgdb_conn.query(sql_desc)
  except Exception, e:
    print "create table failed"
    pgdb_conn.close() 
    return    
   
  #插入记录  
  sql_desc = "INSERT INTO tbl_product3(sv_productname) values('apple')"
  try:
    pgdb_conn.query(sql_desc)
  except Exception, e:
    print "insert record into table failed"
    pgdb_conn.close() 
    return  
    
  #查询表 1    
  sql_desc = "select * from tbl_product3"
  for row in pgdb_conn.query(sql_desc).dictresult():
    print row
 

   
  #关闭数据库连接   
  pgdb_conn.close()    

if __name__ == "__main__": 
  
  #操作数据库
  operate_postgre_tbl_product()

PyGreSQL  with SQLAlchemy 

# coding=utf-8
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# PyGreSQL
eng = create_engine('postgresql+pygresql://mymotif:wxwpxh@127.0.0.1:5432/mymotif')

DB_Session = sessionmaker(bind=eng)
session = DB_Session()
data = session.execute("SELECT * FROM STUDENT")
for row in data:
    for col in row:
        print col,  # python3改为print(col, end=' ')
    print  # python3改为print()

session.close()

参考:http://blog.csdn.net/phashh/article/details/51077943

 

© 著作权归作者所有

wangxuwei
粉丝 26
博文 336
码字总数 119651
作品 0
杭州
其他
私信 提问
Flask Web开发新手指南

Flask框架是Python开发的一个基于Werkzeug和Jinja 2的web开发微框架,它的优势就是极其简洁,但又非常灵活,而且容易学习和应用。因此Flask框架是Python新手快速开始web开发最好的选择,此外...

汇智网教程
2018/12/10
0
0
Airflow 中文文档:初始化数据库后端

如果您想对Airflow进行真正的试驾,您应该考虑设置一个真正的数据库后端并切换到LocalExecutor。 由于Airflow是使用优秀的SqlAlchemy库与其元数据进行交互而构建的,因此您应该能够使用任何支...

ApacheCN_飞龙
2018/11/28
0
0
SQLAlchemy 0.9.0 发布,数据持久层框架

SQLAlchemy 0.9.0 发布,此版本主要更新内容如下: 支持本地 Python 3, 完善了即刻加载系统,包括添加了一个更有表现力的 API 能处理更多复杂的连接 支持 Postgresql JSON 类型,SQL 表达式 ...

oschina
2014/01/01
1K
3
Flask框架web开发:零基础入门

Flask框架是Python开发的一个基于Werkzeug和Jinja 2的web开发微框架,它的优势就是极其简洁,但又非常灵活,而且容易学习和应用。因此Flask框架是Python新手快速开始web开发最好的选择,此外...

笔阁
2018/12/11
0
0
试用PostgreSQL+psycopg2+SQLAlchemy

1.PostgreSQL 因为想添加个gis功能模块,看django推荐gis用PostgreSQL数据库,拿来试用下,安装的时候有几个注意的小问题。 1.To use the yum repository, you must first install the repo...

quanpower
2014/02/27
0
0

没有更多内容

加载失败,请刷新页面

加载更多

Redox OS 发布 0.5 版

Redox OS 是一个几乎完全以 Rust 语言编写的通用操作系统及周围生态(例如文件系统、显示服务器及 Rust 版本的 libc)。其遵循微内核架构,在一定程度上兼容于 POSIX。 该项目于日前发布了 ...

linuxCool
52分钟前
2
0
更新上传git/gitee项目时出现密码配置错误(incorrect username or password)的解决办法

1.输入【git remote add origin git地址】 命令时出现incorrect username or password的错误,这是因为之前弹出输入账户和密码时输错了,而且会一直默认错误的账户和密码,因此需要把之前输入...

west_coast
59分钟前
0
0
Jenkins基础入门-1-Jenkins简单介绍和环境安装

如果在做自动化测试的朋友,应该熟悉Jenkins,或者至少使用过。如果一个人没有使用过Jenkins或者hudson,hudson是Jenkins的前身,他还说自己做过自动化测试,只能说,他只不过是在做半自动化...

shzwork
今天
0
0
linux上解压版安装jdk,tomcat

需要的安装包 1.vmware12 2.centos7版本 3.安装完成后需要xshell来连接远程虚拟机,虚拟机保证要联网,网络畅通。 4.xftp用来向linux传输文件用,一般来说xshell和xftp配套使用 5.对应的压缩...

architect刘源源
今天
29
0
使用 spring 的 IOC 解决程序耦合

工厂模式解耦 在实际开发中我们可以把三层的对象都使用配置文件配置起来,当启动服务器应用加载的时候,让一个类中的方法通过读取配置文件,把这些对象创建出来并存起来。在接下来的使用的时...

骚年锦时
今天
2
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部