PostgreSQL的一个简单ORM。
stellata的Python项目详细描述
stellata是python 3的一个简单的postgresql orm。
连接
使用线程连接到默认主机/端口上的数据库 大小为10的连接池:
import stellata.database db = stellata.database.initialize( name='database', user='user', password='password', host='localhost', port=5432, pool_size=10 )
定义模型
用户模型可能如下所示:
import stellata.fields import stellata.model class User(stellata.model.Model): __table__ = 'users' id = stellata.fields.UUID(null=False) active = stellata.fields.Integer(default=1) name = stellata.fields.Text(null=False) hash = stellata.fields.Varchar(length=255) email = stellata.fields.Text() dt = stellata.fields.Timestamp(null=False)
__table__属性是 PostgreSQL数据库。每个属性对应于一列,并且 stellata.fields中的类用于定义列类型。
字段类型
- 大整数
- 布尔型
- 整数
- 数字
- 文本
- 时间戳
- uuid
- 瓦查尔
关系
和任何好的orm一样,stellata支持模型之间的关系。这是 两个相关的模型,A和B:
import stellata.model import stellata.fields import stellata.relations class A(stellata.model.Model): __table__ = 'a' id = stellata.fields.UUID() foo = stellata.fields.Text() bar = stellata.fields.Integer(default=0, null=False) b = stellata.relations.HasMany(lambda: B.a_id) class B(stellata.model.Model): __table__ = 'b' id = stellata.fields.UUID() a_id = stellata.fields.UUID() bar = stellata.fields.Integer(null=False) a = stellata.relations.BelongsTo(lambda: B.a_id, lambda: A)
指数
索引使查询快速进行。让我们在 表格:
import stellata.model import stellata.fields import stellata.index import stellata.relations class A(stellata.model.Model): __table__ = 'a' id = stellata.fields.UUID() foo = stellata.fields.Text() bar = stellata.fields.Integer(default=0, null=False) b = stellata.relations.HasMany(lambda: B.A_id, lambda: A) primary_key = stellata.index.PrimaryKey(lambda: A.id) foo_index = stellata.index.Index(lambda: A.foo, unique=True)
序列化
一旦你有了一些数据,很快你就会想要转换它 到json。为此,请使用:
stellata.model.serialize(model)
这将递归地序列化对象/关系,您可以传递它 对象、字典、列表等。
meta
在某些情况下,能够遍历所有模型是很方便的。 你已经定义了。例如,您可能希望截断 单元测试。在这种情况下,您可以这样做:
for model in stellata.model.registered(): # do something with model
迁移
一旦定义了模型,就可以通过 执行迁移。
stellata.schema.migrate(db, execute=True)
这里,db是 stellata.database.initialize调用。如果你想干跑步, 不实际执行任何查询,请执行:
stellata.schema.migrate(db)
在这两种情况下,此函数都将返回 移民。
重置
在某些开发脚本中,可能需要清理数据库。如果 如果您愿意,您可以这样做:
stellata.schema.drop_tables_and_lose_all_data(db, execute=True)
顾名思义,这个函数非常具有破坏性,所以不要这样做 在生产数据库上。
积垢操作
最后,让我们看看如何使用stellata查询数据库。
创建
让我们创建一个新的A实例。
a = A.create(A(foo='bar', bar=5)) a.id == '2a12f545-c587-4b99-8fd2-57e79f7c8bca' a.foo == 'bar' a.bar == 4
或者,如果我们想批量创建:
result = A.create([ A(foo='bar', bar=6), A(foo='baz', bar=7) ]) len(result) == 2
如果在某些字段上创建了唯一索引,则可以利用 PostgreSQL冲突处理功能:
A.create(A(foo='baz', bar=9), unique=(A.foo,))
现在,如果已经有一行foo的值为baz, 然后bar列将更新为9值,而不是 而不是创建新行。
读取
要从数据库中读取,我们需要使用where方法。让我们 获取之前创建的A的实例:
a = A.where(A.id == '2a12f545-c587-4b99-8fd2-57e79f7c8bca').get() a.id == '2a12f545-c587-4b99-8fd2-57e79f7c8bca'
天哪,瑞克,那是什么语法?我们在使用运算符重载,莫蒂。 我们还能做什么?
A.where(A.bar < 5).get() A.where(A.bar > 1).get() A.where(A.id << ['2a12f545-c587-4b99-8fd2-57e79f7c8bca', '31be0c81-f5ee-49b9-a624-356402427f76']).get()
最后一个是where-in查询,以防不太明显。我们 也可以在我们的查询中使用and和or,如so:
A.where((A.id == '2a12f545-c587-4b99-8fd2-57e79f7c8bca') | (A.bar < 5)).get() A.where((A.id == '2a12f545-c587-4b99-8fd2-57e79f7c8bca') & (A.bar > 1)).get()
其他铃声和口哨声:
A.where(A.bar < 5).order(A.bar, 'asc').limit(5).get()
常见的读取操作是查找列与某些行匹配的所有行 值,因此我们可以使用速记:
A.find('2a12f545-c587-4b99-8fd2-57e79f7c8bca')
默认情况下,将使用id字段,但也可以指定 自己的字段:
A.find(5, A.bar)
如果给定一个列表,find将返回一个根据 指定的字段:
a = A.find(['2a12f545-c587-4b99-8fd2-57e79f7c8bca', '31be0c81-f5ee-49b9-a624-356402427f76']) a['2a12f545-c587-4b99-8fd2-57e79f7c8bca'].id == '2a12f545-c587-4b99-8fd2-57e79f7c8bca' a['31be0c81-f5ee-49b9-a624-356402427f76'].id == '31be0c81-f5ee-49b9-a624-356402427f76'
连接
我们可以使用我们之前通过连接建立的那些关系。假设我们 创建以下内容:
a = A.create(A(foo='bar', bar=5)) a.id == '2a12f545-c587-4b99-8fd2-57e79f7c8bca' b = B.create([ B(a_id='2a12f545-c587-4b99-8fd2-57e79f7c8bca', qux=3) B(a_id='2a12f545-c587-4b99-8fd2-57e79f7c8bca', qux=5) ])
现在,我们可以这样做:
a = A.join(A.b).where(A.id == '2a12f545-c587-4b99-8fd2-57e79f7c8bca').get() a.id == '2a12f545-c587-4b99-8fd2-57e79f7c8bca' len(a.b) == 2 a.b[0].qux == 3 a.b[1].qux == 5
或者,另一种方式:
b = B.join(B.a).where(B.qux << [3, 5]).get() len(b) == 2 b[0].qux == 3 b[0].a.id == '2a12f545-c587-4b99-8fd2-57e79f7c8bca' b[1].qux == 5 b[1].a.id == '2a12f545-c587-4b99-8fd2-57e79f7c8bca'
默认情况下,连接将通过多个选择查询执行。如果你愿意 更愿意加入,请执行以下操作:
a = A.join_with('join').join(A.b).where(A.id == '2a12f545-c587-4b99-8fd2-57e79f7c8bca').get()
结果与以前相同,但基础查询是 差异埃伦特。你用哪种方法完全取决于你,而且可能因 不同的问题。
更新
如您所料,update查询将创建和 阅读:
A.where(A.id == '2a12f545-c587-4b99-8fd2-57e79f7c8bca').update(A(bar=7))
删除
这个现在很容易。
A.where(id == '2a12f545-c587-4b99-8fd2-57e79f7c8bca').delete()
或者,如果您喜欢单个截断操作:
A.truncate()
交易
Stellata还支持PostgreSQL事务:
A.begin() A.truncate() A.create([A(bar=1), A(bar=2)]) A.commit()