orm 基础 - orator 的常用操作
记录一些自己使用 orator 的常用操作
Mysql建表，建用户操作 create database xxx CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
CREATE USER 'username'@'%' IDENTIFIED BY 'new_password';
grant all privileges on database to 'usename'@'%';MySQL基本配置 from orator import DatabaseManager
config = {
'mysql': {
'driver': 'mysql',
'host': 'localhost',
'database': 'database',
'user': 'root',
'password': '',
'prefix': ''
}
}
db = DatabaseManager(config)多db连接 有default: ‘mysql’(这种字符串是自定的) 或者有一个 default 的连接 在 Model 上设置 `__connection__` 即可 实际的 Model上不需要再使用 on from orator import DatabaseManager, Schema
databases = {
'default': 'mysql',
'mysql': {
'driver': 'mysql',
'host': '127.0.0.1',
'database': 'spider_zditect_com',
'user': 'root',
'password': '123456',
'prefix': ''
},
'txz_bloger': {
'driver': 'mysql',
'host': '127.0.0.1',
'database': 'txz_zbloger',
'user': 'root',
'password': '123456',
'prefix': ''
},
}
db = DatabaseManager(databases)
schema = Schema(db)# 连接 default
Entry.where({'is_crawled': False}).get()
# 连接名称为: txz_bloger 的数据库
SpiderPost.on('txz_bloger').limit(1).get()另一种，多数据库的操作 from .abstract_model import AbstractModel
class SpiderPost(AbstractModel):
__connection__ = 'txz_bloger'
__table__ = 'spider_posts'对 model 的操作 # 创建 model
orator make:model User
# 创建 model 和 migration
orator make:model User -m对 migration 的操作 # 默认位置创建
orator make:migration create_users_table
# 指定 path 创建
orator make:migration create_users_table -p my/path/to/migrations一个db.py from orator import DatabaseManager, Schema
databases = {
'mysql': {
'driver': 'mysql',
'host': '127.0.0.1',
'database': 'gbins_spider',
'user': 'root',
'password': '123456',
'prefix': ''
}
}
db = DatabaseManager(databases)
schema = Schema(db)Run 一个 migration cd ./scrapy-notes/src/2022/01/myspider/myspider
orator migrate --config ../orator.yml
orator migrate --config ./db.py管理 migration 运行的表 为某个字段添加索引 table.index('fasta_id')关于循环引用的问题 (most likely due to a circular import
chunks 原理 # Split a Python List into Chunks using For Loops
our_list = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]
chunked_list = list()
chunk_size = 3
for i in range(0, len(our_list), chunk_size):
chunked_list.append(our_list[i:i+chunk_size])
print(chunked_list)
# Returns: [[1, 2, 3], [4, 5, 6], [7, 8, 9], [10, 11]]时区问题设置 from orator import Model
import pendulum
class AbstractModel(Model):
# normalize timezone
def fresh_timestamp(self):
return pendulum.now('Asia/Shanghai')
常用操作 操作 代码 备注 根据条件查询 where # 根据条件得到结果
records = Url.where({"quality": 'low'}).get()可用 for..in 来遍历 针对 NULL 的情况 # 查询空的情况
Url.where_null('category').where({"is_crawled": False})
# 查询不为空的情况
Url.where_not_null('pmid').where({"is_crawled": False}) 取得符合条件的前10条 users = User.where('votes', '>', 100).take(10).get()
for user in users:
print(user.name) 根据id查询 # 查询1条结果
user = User.find(1)取得单条record 根据其它条件查询 # 与 rails 里的 find_by
Fasta.where({"url_id": 1}).first()注意.first() 计算属性 from orator.orm import Model, accessor
class Entry(Model):
@accessor
def xml(self):
return f"https://www.uniprot.org/uniprot/{self.entry_id}.xml"用这个定义 @accessor 取得 chunks # 直接遍历 chunks
chunks = User.chunk(100)
for users in chunks:
for user in users:
# ...
# 也可以转为2维list再完成其它操作
list2 = list(chunks)好处是分段处理 根据like关键字 records = MyModel.query().where('region_name', 'like', '%keyword%').get()
for entity in records:
print("entity: ", entity)
Migration 常用 操作 代码 备注 id table.increments('id')自增id timestamp table.timestamps()created_at / updated_at enum table.enum('quality', ['low', 'medium', 'high'])枚举型字段 boolean table.boolean('is_crawled').default(False)布尔类型，默认为 false 创建索引 # 联合唯一
table.unique(['seq_id', 'url_id'])
# 自己定义索引名称
t.unique(['field_with_really_long_name', 'another_field_with_long_name'], name='my_uniq_idx')
# 普通
table.index('url_id')适合多字段组成的唯一索引
参考
