首页 热点资讯 义务教育 高等教育 出国留学 考研考公
您的当前位置:首页正文

SQLAlchemy 操作数据库

2023-11-10 来源:花图问答

pip3.4 install PyMySQLsudo pip3.4 install sqlalchemy

代码:

 1 #!/usr/bin/env python 2 # encoding: utf-8 3 """ 4 @author: 侠之大者kamil 5 @file: Sqlalchemy_test1.py 6 @time: 16/4/11 下午10:41 7 """ 8 #from sqlalchemy import create_engine,Table,column,Integer,ForeignKey 9 from sqlalchemy import *10 11 metadata = MetaData()12 13 user = Table("user",metadata,14 Column(‘id‘,Integer,primary_key=True),15 Column(‘name‘,String(20)),16 )17 18 color = Table("color",metadata,19 Column(‘GUID‘,Integer,primary_key=True),20 Column(‘code‘,String(20)),)21 22 engine = create_engine("mysql+pymysql://myuser:lbk1992@xxxx.hk/python_one",max_overflow=5)23 #python3 没有 mysqldb 支持,在此使用 pymysql.24 metadata.create_all(engine)25 26 metadata.clear()

 

SQLAlchemy 操作数据库

标签:

小编还为您整理了以下内容,可能对您也有帮助:

在python3下怎样用flask-sqlalchemy对mysql数据库操作

以 Debian/Ubuntu 为例(请确保有管理员权限):

1.MySQL

代码如下:

apt-get install mysql-server

apt-get install mysql-client

apt-get install libmysqlclient15-dev

2.python-mysqldb

代码如下:

apt-get install python-mysqldb

3.easy_install

代码如下:

wget http:// peak.telecommunity.com/dist/ez_setup.py

python ez_setup.py

4.MySQL-Python

代码如下:

easy_install MySQL-Python

5.SQLAlchemy

代码如下:

easy_install SQLAlchemy

6、安装完成后使用下面代码测试连接

代码如下:

from sqlalchemy import create_engine

from sqlalchemy.orm import sessionmaker

DB_CONNECT_STRING = 'mysql+mysqldb://root:123@localhost/ooxx?charset=utf8'

engine = create_engine(DB_CONNECT_STRING, echo=True)

DB_Session = sessionmaker(bind=engine)

session = DB_Session()

7、数据操作(增删改查)

代码如下:

from sqlalchemy import func, or_, not_

user = User(name='a')

session.add(user)

user = User(name='b')

session.add(user)

user = User(name='a')

session.add(user)

user = User()

session.add(user)

session.commit()

query = session.query(User)

print query # 显示SQL 语句

print query.statement # 同上

for user in query: # 遍历时查询

print user.name

print query.all() # 返回的是一个类似列表的对象

print query.first().name # 记录不存在时,first() 会返回 None

# print query.one().name # 不存在,或有多行记录时会抛出异常

print query.filter(User.id == 2).first().name

print query.get(2).name # 以主键获取,等效于上句

print query.filter('id = 2').first().name # 支持字符串

query2 = session.query(User.name)

print query2.all() # 每行是个元组

print query2.limit(1).all() # 最多返回 1 条记录

print query2.offset(1).all() # 从第 2 条记录开始返回

print query2.order_by(User.name).all()

print query2.order_by('name').all()

print query2.order_by(User.name.desc()).all()

print query2.order_by('name desc').all()

print session.query(User.id).order_by(User.name.desc(), User.id).all()

print query2.filter(User.id == 1).scalar() # 如果有记录,返回第一条记录的第一个元素

print session.query('id').select_from(User).filter('id = 1').scalar()

print query2.filter(User.id > 1, User.name != 'a').scalar() # and

query3 = query2.filter(User.id > 1) # 多次拼接的 filter 也是 and

query3 = query3.filter(User.name != 'a')

print query3.scalar()

print query2.filter(or_(User.id == 1, User.id == 2)).all() # or

print query2.filter(User.id.in_((1, 2))).all() # in

query4 = session.query(User.id)

print query4.filter(User.name == None).scalar()

print query4.filter('name is null').scalar()

print query4.filter(not_(User.name == None)).all() # not

print query4.filter(User.name != None).all()

print query4.count()

print session.query(func.count('*')).select_from(User).scalar()

print session.query(func.count('1')).select_from(User).scalar()

print session.query(func.count(User.id)).scalar()

print session.query(func.count('*')).filter(User.id > 0).scalar() # filter() 中包含 User,因此不需要指定表

print session.query(func.count('*')).filter(User.name == 'a').limit(1).scalar() == 1 # 可以用 limit() count() 的返回数

print session.query(func.sum(User.id)).scalar()

print session.query(func.now()).scalar() # func 后可以跟任意函数名,只要该数据库支持

print session.query(func.current_timestamp()).scalar()

print session.query(func.md5(User.name)).filter(User.id == 1).scalar()

query.filter(User.id == 1).update({User.name: 'c'})

user = query.get(1)

print user.name

user.name = 'd'

session.flush() # 写数据库,但并不提交

print query.get(1).name

session.delete(user)

session.flush()

print query.get(1)

session.rollback()

print query.get(1).name

query.filter(User.id == 1).delete()

session.commit()

print query.get(1)

在python3下怎样用flask-sqlalchemy对mysql数据库操作

这个问题经常难道新手一下,因为大部分教程里(包括经典的《Flask

Web开发》一书),告诉了我们如何使用flask-sqlalchemy操作sqlite,但在生产环境(线上网站)上,我们肯定是使用MySQL或其

他,而大部分的教程里,又告诉我们flask-sqlalchemy使用MySQL的方式是:

mysql://username:password@server/db

结果我们照葫芦画瓢的来一下,发现压根不行,写好的网站一跟数据库沾边就报错。

Python和MySQL是「两个国家的人」,他们互不相通,因而需要一个中间代理,让双方互通有无,跟翻译一样(这比喻不准确,但足够你明白意思就行)。翻译又有很多选择,不同的翻译各有特色。

主解决问题选择的翻译是「flask-mysqldb」,其背后的主子是「MySQL-python」。恩,说到这里你应该知道,「flask-xxx」

这样的包都是对背后主子进行了适合Flask封装的插件,跟包子皮一样,里面的馅才是重点,「flask-mysqldb」的馅是「MySQL-

python」。

而我要推荐的是另一个翻译:PyMySQL,这玩意的好处是可以做异步(「MySQL-python」也可以,个人口味罢了),简而言之,网站访问量大了就需要考虑异步,现在别管这是啥子。这玩意的安装方式是:

pip install PyMySQL

之后,数据库连接由:

mysql://username:password@server/db

改为

mysql+pymysql://username:password@server/db

就可以了。

在python3下怎样用flask-sqlalchemy对mysql数据库操作

MHA监控复制架构的主服务器,一旦检测到主服务器故障,就会自动进行故障转移。即使有些从服务器没有收到最新的relay log,MHA自动从最新的从服务器上识别差异的relay log并把这些日志应用到其他从服务器上,因此所有的从服务器保持一致性了。MHA通常在几秒内完成故障转移,9-12秒可以检测出主服务器故障,7-10秒内关闭故障的主服务器以避免脑裂,几秒中内应用差异的relay log到新的主服务器上,整个过程可以在10-30s内完成。还可以设置优先级指定其中的一台slave作为master的候选人。由于MHA在slaves之间修复一致性,因此可以将任何slave变成新的master,而不会发生一致性的问题,从而导致复制失败。

2. 交互式主服务器故障转移

可以只使用MHA的故障转移,而不用于监控主服务器,当主服务器故障时,人工调用MHA来进行故障故障。

3. 非交互式的主故障转移

不监控主服务器,但自动实现故障转移。这种特征适用于已经使用其他软件来监控主服务器状态,比如heartbeat来检测主服务器故障和虚拟IP地址接管,可以使用MHA来实现故障转移和slave服务器晋级为master服务器。

4. 在线切换主从服务器

在许多情况下,需要将现有的主服务器迁移到另外一台服务器上。比如主服务器硬件故障,RAID控制卡需要重建,将主服务器移到性能更好的服务器上等等。维护主服务器引起性能下降,导致停机时间至少无法写入数据。另外,阻塞或杀掉当前运行的会话会导致主主之间数据不一致的问题发生。MHA提供快速切换和优雅的阻塞写入,这个切换过程只需要0.5-2s的时间,这段时间内数据是无法写入的。在很多情况下,0.5-2s的阻塞写入是可以接受的。因此切换主服务器不需要计划分配维护时间窗口(呵呵,不需要你在夜黑风高时通宵达旦完成切换主服务器的任务)。

5.MHA由两部分组成:MHA Manager(管理节点)和MHA Node(数据节点)

要搭建MHA,要求一个复制集群中必须最少有三台数据库服务器,一主二从,即一台充当master,一台充当备用master,另外一台充当从库,管理节点可以和master在一台机器上。所以如果你只有二台机器的话,heartbeat,keepalive等都是不错的选择了。

6.MHA比较灵活,可以写脚本,来进行故障转移,或者主从切换等。

在python3下怎样用flask-sqlalchemy对mysql数据库操作

如果你了解SQLite或其它数据驱动的操作方式,只需要改变连接地址字符串属性即可

app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://user:password@localhost/mydatabase'

注意替换真实的账号/密码/地址/数据库名.

官方文档地址: http://flask-sqlalchemy.pocoo.org/2.1/

显示全文