python之sqlalchemy创建表的实例详解

来源:脚本之家  责任编辑:小易  

假设你已经安装了sqlalchemy和mysql数据库1、Python中安装pymysql3

python之sqlalchemy创建表的实例详解

我觉得你问问题的方式不对,应该是python使用sqlalchemy连接数据库,不是连接python

通过sqlalchemy创建表需要三要素:引擎,基类,元素

唔,貌似题主问题已经解决了,挺好的。我还是答一下,给后续需要的新手一点帮助。这个问题经常难道

没用过flask_sqlalchemy,按照你的思路sql应该是这样的, 按创建时间升序查找第一个s

from sqlalchemy import create_engine 
from sqlalchemy.ext.declarative import declarative_base 
from sqlalchemy import Column,Integer,String 

对于in 和 exists的区别: 如果子查询得出的结果集记录较少,主查询中的表较大且又有索引时应该

引擎:也就是实体数据库连接

买车这个事情仁者见仁智者见智,爱买什么买什么,没必要争。自己喜欢什么买什么就好了,没必要强迫他人一定要跟随自己的想法来。最后再三强调一句,二手车,最重要的永远不是价格,而是车况、车况、车况,重要的事情说三遍。不管你信任不信任现在的第三方二手车平台,反正这些平台每天都会无数次被车贩子骚扰。因为这些贩子都想从这里面收车,足见这些平台里放出来的车价格是如何具有诱惑力。01.2008年桑塔纳志俊1.8L手动油耗低平均7L空间超级大直列4缸噪音大02.2008款飞度耐用省油小型车的最大空间保值3-4万买近年份低配手动1.3L或年份久的高配自动1.5L03.高尔夫MK41.8L手动档整车结实,高速180码

没用过flask_sqlalchemy,按照你的思路sql应该是这样的, 按创建时间升序查找第一个s

engine = create_engine('mysql+pymysql://godme:godme@localhost/godme',encoding='utf-8',echo=True)

今天我要说的是关于本人的一个故事,一个关于《魔兽世界》JJC的故事个人认为魔兽世界PVP做的算是非常平衡了,首先天赋百花齐放,百家争鸣。部落最良心的地方就是不会出现土豪吊打屌丝的这种情况出现,只要你手法好,预判好,照样可以战胜装备比你好的玩家。话不多说,下面来看看我的JJC魔兽生涯!(本人S3赛季末刚到70级德鲁伊)一个来自部落的牛头德鲁伊依稀的记得当初第一次打JJC的时候,我们组了一个所谓的便当队,带着打副本的心态进入了洛丹伦废墟,我方战士当时用语音喊到:兄弟们我上去了~群嘲!!!吼了三个快输出!!不到三秒就躺地,哈哈哈每每想到这个画面我还是会忍不住笑出声!(玩过魔兽世界的人都知道嘲讽是只能

传入参数:数据库类型+连接库+用户名+密码+主机,字符编码,是否打印建表细节

宇宙这个词语,从字面上可以看到,它有2个字组成,其中“宇”就是空间,“宙”就是时间。因此,宇宙就是空间加上时间的一个组合体,其实也可以认为是整个时空。当然了,宇宙大爆炸的瞬间是时间开始的地方。问题的关键在于,其实时间与空间都是依赖于人的,我们人可以对时空做不同的3+1分解,得到时间与空间。一般正常的3+1分解是这样的,我们把时空分解成3维空间与1维时间,其中时间方向规定为我们人类的固有时间的方向。因为我们人类的运动速度都差不多,所以近似可以认为,我们人类拥有一个公共的时间方向。到了未来社会,一旦发明了近光速的飞船等星际旅行的交通工具,人与人之间的时间方向不再近似重合,那时候——不同人会对宇宙做

基类:

没用过flask_sqlalchemy,按照你的思路sql应该是这样的, 按创建时间升序查找第一个s

Base = declarative_base()

元素:

没用过flask_sqlalchemy,按照你的思路sql应该是这样的, 按创建时间升序查找第一个s

class User(Base): __tablename__ = 'user' id = Column(Integer,primary_key=True) name = Column(String(32)) password = Column(String(64))

通过基本元素:

没用过flask_sqlalchemy,按照你的思路sql应该是这样的, 按创建时间升序查找第一个s

__tablename__:指定表名 Column:行声明,可指定主键 Integer:数据类型 String:数据类型,可指定长度

创建:

没用过flask_sqlalchemy,按照你的思路sql应该是这样的, 按创建时间升序查找第一个s

Base.metadata.create_all(engine)

基本过程:

1. 获取实体数据库连接

2. 创建类,继承基类,用基本类型描述数据库结构

3. 基类调用类结构,根据描述在引擎上创建数据表

如有疑问请留言或者到本站社区交流讨论,感谢阅读,希望能帮助到大家,谢谢大家对本站的支持!

扩展阅读,根据您访问的内容系统为您准备了以下内容,希望对您有帮助。

Python-SQLAlchemy中,定义表对象时,何时应该写

描述问题

从Base继承的表对象, 何时写__init__? (标准是啥?)

我自己的观点是: 貌似写__init__完全没有必要

上下文环境

Python3

SQLAlchemy

Metadata-Version: 1.1Name: SQLAlchemyVersion: 1.0.11Summary: Database Abstraction Library

重现

相关代码

"""model.py

The datamodel, which represents Person that has multiple

Address objects, each with PostalCode, City, Country.

Person --(1..n)--> Address

Address --(has a)--> PostalCode

PostalCode --(has a)--> City

City --(has a)--> Country

"""from sqlalchemy import Column, Integer, String, ForeignKeyfrom sqlalchemy.orm import relationshipfrom .caching_query import FromCache, RelationshipCachefrom .environment import Base, bootstrapclass Country(Base):

__tablename__ = 'country'

id = Column(Integer, primary_key=True)

name = Column(String(100), nullable=False) def __init__(self, name):

self.name = nameclass City(Base):

__tablename__ = 'city'

id = Column(Integer, primary_key=True)

name = Column(String(100), nullable=False)

country_id = Column(Integer, ForeignKey('country.id'), nullable=False)

country = relationship(Country) def __init__(self, name, country):

self.name = name

self.country = countryclass PostalCode(Base):

__tablename__ = 'postal_code'

id = Column(Integer, primary_key=True)

code = Column(String(10), nullable=False)

city_id = Column(Integer, ForeignKey('city.id'), nullable=False)

city = relationship(City) @property

def country(self):

return self.city.country def __init__(self, code, city):

self.code = code

self.city = cityclass Address(Base):

__tablename__ = 'address'

id = Column(Integer, primary_key=True)

person_id = Column(Integer, ForeignKey('person.id'), nullable=False)

street = Column(String(200), nullable=False)

postal_code_id = Column(Integer, ForeignKey('postal_code.id'))

postal_code = relationship(PostalCode) @property

def city(self):

return self.postal_code.city @property

def country(self):

return self.postal_code.country def __str__(self):

return "%s\t"\ "%s, %s\t"\ "%s" % (self.street, self.city.name,

self.postal_code.code, self.country.name)class Person(Base):

__tablename__ = 'person'

id = Column(Integer, primary_key=True)

name = Column(String(100), nullable=False)

addresses = relationship(Address, collection_class=set) def __init__(self, name, *addresses):

self.name = name

self.addresses = set(addresses) def __str__(self):

return self.name def __repr__(self):

return "Person(name=%r)" % self.name def format_full(self):

return "\t".join([str(x) for x in [self] + list(self.addresses)])# Caching options. A set of three RelationshipCache options# which can be applied to Query(), causing the "lazy load"# of these attributes to be loaded from cache.cache_address_bits = RelationshipCache(PostalCode.city, "default").\

and_(

RelationshipCache(City.country, "default")

).and_(

RelationshipCache(Address.postal_code, "default")

)

bootstrap()

报错信息

相关截图

已经尝试哪些方法仍然没有解决(附上相关链接)

问题简化

python sqlalchemy orm中,怎样往已有数据的mysql表插入一列数据

看你截图需求本意应该是需要把orderID和totalAmount一起拿到列表里循环更新表数据、不存在插入一列数据的概念。

python sqlalchemy 多线程怎么写

首先说下,由于最新的 0.8 版还是开发版本,因此我使用的是 0.79 版,API 也许会有些不同。

因为我是搭配 MySQL InnoDB 使用,所以使用其他数据库的也不能完全照搬本文。

接着就从安装开始介绍吧,以 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

python ez_setup.py

4.MySQL-Python

复制代码代码如下:

easy_install MySQL-Python

5.SQLAlchemy

复制代码代码如下:

easy_install SQLAlchemy

如果是用其他操作系统,遇到问题就 Google 一下吧。我是在 Mac OS X 上开发的,途中也遇到些问题,不过当时没记下来……

值得一提的是我用了 MySQL-Python 来连 MySQL,因为不支持异步调用,所以和 Tornado 不是很搭。不过性能其实很好,因此以后再去研究下其他方案吧……

装好后就可以开始使用了:

复制代码代码如下:

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()

这里的 DB_CONNECT_STRING 就是连接数据库的路径。“mysql+mysqldb”指定了使用 MySQL-Python 来连接,“root”和“123”分别是用户名和密码,“localhost”是数据库的域名,“ooxx”是使用的数据库名(可省略),“charset”指定了连接时使用的字符集(可省略)。

create_engine() 会返回一个数据库引擎,echo 参数为 True 时,会显示每条执行的 SQL 语句,生产环境下可关闭。

sessionmaker() 会生成一个数据库会话类。这个类的实例可以当成一个数据库连接,它同时还记录了一些查询的数据,并决定什么时候执行 SQL 语句。由于 SQLAlchemy 自己维护了一个数据库连接池(默认 5 个连接),因此初始化一个会话的开销并不大。对 Tornado 而言,可以在 BaseHandler 的 initialize() 里初始化:

复制代码代码如下:

class BaseHandler(tornado.web.RequestHandler):

def initialize(self):

self.session = models.DB_Session()

def on_finish(self):

self.session.close()

对其他 Web 服务器来说,可以使用 sqlalchemy.orm.scoped_session,它能保证每个线程获得的 session 对象都是唯一的。不过 Tornado 本身就是单线程的,如果使用了异步方式,就可能会出现问题,因此我并没使用它。

拿到 session 后,就可以执行 SQL 了:

复制代码代码如下:

session.execute('create database abc')

print session.execute('show databases').fetchall()

session.execute('use abc')

# 建 user 表的过程略

print session.execute('select * from user where id = 1').first()

print session.execute('select * from user where id = :id', {'id': 1}).first()

不过这和直接使用 MySQL-Python 没啥区别,所以就不介绍了;我还是喜欢 ORM 的方式,这也是我采用 SQLAlchemy 的唯一原因。

于是来定义一个表:

复制代码代码如下:

from sqlalchemy import Column

from sqlalchemy.types import CHAR, Integer, String

from sqlalchemy.ext.declarative import declarative_base

BaseModel = declarative_base()

def init_db():

BaseModel.metadata.create_all(engine)

def drop_db():

BaseModel.metadata.drop_all(engine)

class User(BaseModel):

__tablename__ = 'user'

id = Column(Integer, primary_key=True)

name = Column(CHAR(30)) # or Column(String(30))

init_db()

declarative_base() 创建了一个 BaseModel 类,这个类的子类可以自动与一个表关联。

以 User 类为例,它的 __tablename__ 属性就是数据库中该表的名称,它有 id 和 name 这两个字段,分别为整型和 30 个定长字符。Column 还有一些其他的参数,我就不解释了。

最后,BaseModel.metadata.create_all(engine) 会找到 BaseModel 的所有子类,并在数据库中建立这些表;drop_all() 则是删除这些表。

接着就开始使用这个表吧:

复制代码代码如下:

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)

增删改查都涉及到了,自己看看输出的 SQL 语句就知道了,于是基础知识就介绍到此了。

下面开始介绍一些进阶的知识。

如何批量插入大批数据?

可以使用非 ORM 的方式:

复制代码代码如下:

session.execute(

User.__table__.insert(),

[{'name': `randint(1, 100)`,'age': randint(1, 100)} for i in xrange(10000)]

)

session.commit()

上面我批量插入了 10000 条记录,半秒内就执行完了;而 ORM 方式会花掉很长时间。

如何让执行的 SQL 语句增加前缀?

使用 query 对象的 prefix_with() 方法:

复制代码代码如下:

session.query(User.name).prefix_with('HIGH_PRIORITY').all()

session.execute(User.__table__.insert().prefix_with('IGNORE'), {'id': 1, 'name': '1'})

如何替换一个已有主键的记录?

使用 session.merge() 方法替代 session.add(),其实就是 SELECT + UPDATE:

复制代码代码如下:

user = User(id=1, name='ooxx')

session.merge(user)

session.commit()

或者使用 MySQL 的 INSERT … ON DUPLICATE KEY UPDATE,需要用到 @compiles 装饰器,有点难懂,自己搜索看吧:《SQLAlchemy ON DUPLICATE KEY UPDATE》 和 sqlalchemy_mysql_ext。

如何使用无符号整数?

可以使用 MySQL 的方言:

复制代码代码如下:

from sqlalchemy.dialects.mysql import INTEGER

id = Column(INTEGER(unsigned=True), primary_key=True)

模型的属性名需要和表的字段名不一样怎么办?

开发时遇到过一个奇怪的需求,有个其他系统的表里包含了一个“from”字段,这在 Python 里是关键字,于是只能这样处理了:

复制代码代码如下:

from_ = Column('from', CHAR(10))

如何获取字段的长度?

Column 会生成一个很复杂的对象,想获取长度比较麻烦,这里以 User.name 为例:

复制代码代码如下:

User.name.property.columns[0].type.length

如何指定使用 InnoDB,以及使用 UTF-8 编码?

最简单的方式就是修改数据库的默认配置。如果非要在代码里指定的话,可以这样:

复制代码代码如下:

class User(BaseModel):

__table_args__ = {

'mysql_engine': 'InnoDB',

'mysql_charset': 'utf8'

}

MySQL 5.5 开始支持存储 4 字节的 UTF-8 编码的字符了,iOS 里自带的 emoji(如 �� 字符)就属于这种。

如果是对表来设置的话,可以把上面代码中的 utf8 改成 utf8mb4,DB_CONNECT_STRING 里的 charset 也这样更改。

如果对库或字段来设置,则还是自己写 SQL 语句比较方便,具体细节可参考《How to support full Unicode in MySQL databases》。

不建议全用 utf8mb4 代替 utf8,因为前者更慢,索引会占用更多空间。

如何设置外键约束?

复制代码代码如下:

from random import randint

from sqlalchemy import ForeignKey

class User(BaseModel):

__tablename__ = 'user'

id = Column(Integer, primary_key=True)

age = Column(Integer)

class Friendship(BaseModel):

__tablename__ = 'friendship'

id = Column(Integer, primary_key=True)

user_id1 = Column(Integer, ForeignKey('user.id'))

user_id2 = Column(Integer, ForeignKey('user.id'))

for i in xrange(100):

session.add(User(age=randint(1, 100)))

session.flush() # 或 session.commit(),执行完后,user 对象的 id 属性才可以访问(因为 id 是自增的)

for i in xrange(100):

session.add(Friendship(user_id1=randint(1, 100), user_id2=randint(1, 100)))

session.commit()

session.query(User).filter(User.age < 50).delete()

执行这段代码时,你应该会遇到一个错误:

复制代码代码如下:

sqlalchemy.exc.IntegrityError: (IntegrityError) (1451, 'Cannot delete or update a parent row: a foreign key constraint fails (`ooxx`.`friendship`, CONSTRAINT `friendship_ibfk_1` FOREIGN KEY (`user_id1`) REFERENCES `user` (`id`))') 'DELETE FROM user WHERE user.age < %s' (50,)

原因是删除 user 表的数据,可能会导致 friendship 的外键不指向一个真实存在的记录。在默认情况下,MySQL 会拒绝这种操作,也就是 RESTRICT。InnoDB 还允许指定 ON DELETE 为 CASCADE 和 SET NULL,前者会删除 friendship 中无效的记录,后者会将这些记录的外键设为 NULL。

除了删除,还有可能更改主键,这也会导致 friendship 的外键失效。于是相应的就有 ON UPDATE 了。其中 CASCADE 变成了更新相应的外键,而不是删除。

而在 SQLAlchemy 中是这样处理的:

复制代码代码如下:

class Friendship(BaseModel):

__tablename__ = 'friendship'

id = Column(Integer, primary_key=True)

user_id1 = Column(Integer, ForeignKey('user.id', ondelete='CASCADE', onupdate='CASCADE'))

user_id2 = Column(Integer, ForeignKey('user.id', ondelete='CASCADE', onupdate='CASCADE'))

如何连接表?

复制代码代码如下:

from sqlalchemy import distinct

from sqlalchemy.orm import aliased

Friend = aliased(User, name='Friend')

print session.query(User.id).join(Friendship, User.id == Friendship.user_id1).all() # 所有有朋友的用户

print session.query(distinct(User.id)).join(Friendship, User.id == Friendship.user_id1).all() # 所有有朋友的用户(去掉重复的)

print session.query(User.id).join(Friendship, User.id == Friendship.user_id1).distinct().all() # 同上

print session.query(Friendship.user_id2).join(User, User.id == Friendship.user_id1).order_by(Friendship.user_id2).distinct().all() # 所有被别人当成朋友的用户

print session.query(Friendship.user_id2).select_from(User).join(Friendship, User.id == Friendship.user_id1).order_by(Friendship.user_id2).distinct().all() # 同上,join 的方向相反,但因为不是 STRAIGHT_JOIN,所以 MySQL 可以自己选择顺序

print session.query(User.id, Friendship.user_id2).join(Friendship, User.id == Friendship.user_id1).all() # 用户及其朋友

print session.query(User.id, Friendship.user_id2).join(Friendship, User.id == Friendship.user_id1).filter(User.id < 10).all() # id 小于 10 的用户及其朋友

print session.query(User.id, Friend.id).join(Friendship, User.id == Friendship.user_id1).join(Friend, Friend.id == Friendship.user_id2).all() # 两次 join,由于使用到相同的表,因此需要别名

print session.query(User.id, Friendship.user_id2).outerjoin(Friendship, User.id

python sqlalchemy 查询条件怎么写

  通过Flask-SQLAlchemy提供的一个query属性,当你通过model类的query属性,你可以得到一个数据库表的查询结果集。

  i.User.query.filter_by(username='peter').first(),通过filter_by方法里的条件表达式来对query所得到的结果集进行过滤,得到你想要得到的结果。

   example:

    Retrieve a user by username通过username属性为’peter‘过滤结果集:

    >>> peter = User.query.filter_by(username='peter').first()

    >>> peter.id

    1

    >>> peter.email

    u'peter@example.org'

   当不存在结果集时返回none:

    >>> missing = User.query.filter_by(username='missing').first()

    >>> missing is None

    True 

python sqlalchemy 怎么用原生sql语句

接着就从安装开始介绍吧,以 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 eak.telecommunity.com/dist/ez_setup.py

python ez_setup.py

4.MySQL-Python

复制代码代码如下:

easy_install MySQL-P

  • 本文相关:
  • python的sqlalchemy模块连接与操作mysql的基础示例
  • python sqlalchemy基本操作和常用技巧(包含大量实例,非常好)
  • 浅析python中sqlalchemy排序的一个坑
  • python程序中使用sqlalchemy时出现乱码的解决方案
  • python sqlalchemy动态添加数据表字段实例解析
  • python的sqlalchemy框架使用入门
  • python 数据库操作 sqlalchemy的示例代码
  • python使用sqlalchemy模块连接数据库操作示例
  • sqlite教程(十三):c语言编程实例代码(1)
  • sqlite3 命令行操作指南
  • sqlite3 api 编程手册
  • sqlite中重置自动编号列的方法
  • sqlite数据库常用语句及mac上的sqlite可视化工具meassqllite使用
  • sqlite教程(十一):临时文件
  • 基于sqlite特殊字符转义的实现方法
  • sqlite优化方法
  • python之sqlalchemy创建表的实例详解
  • sqlite教程(八):命令行工具介绍
  • Python-SQLAlchemy中,定义表对象时,何时应该写
  • python sqlalchemy orm中,怎样往已有数据的mysql表插入一列数据
  • python sqlalchemy 多线程怎么写
  • python sqlalchemy 查询条件怎么写
  • python sqlalchemy 怎么用原生sql语句
  • 在python3下怎样用flask-sqlalchemy对mysql数据库操作
  • 如何使用ORM中的网红“sqlalchemy”,连接python数据库
  • 在python3下怎样用flask-sqlalchemy对mysql数据库操作
  • 请问:python3 sqlalchemy query循环查询结果不变
  • Python 使用原生 SQL 或者 SQLAlchemy,查询效率上差别大吗
  • 免责声明 - 关于我们 - 联系我们 - 广告联系 - 友情链接 - 帮助中心 - 频道导航
    Copyright © 2017 www.zgxue.com All Rights Reserved