terminate.jpg

首页

技术分享

自定义字段类型并自动写文件

首先定义自定义字段类型, 如果需要进一步定制化,可以实现自己的Storage类,只要重新_save 方法就好,可以存S3或者各种网络fs。

myfields.py

import os
from sqlalchemy.types import UserDefinedType

# 默认存储类,如果有定制需求可以自定义
class Storage(object):
def save(self, name, content):
return self._save(name, content)

def _save(self, name, content):
with open(name, 'wb') as fp:
fp.write(content.read())


# 自定义字段类型
class MyField(UserDefinedType):
def __init__(self, storage=Storage(), upload_to=''):
self.storage = storage
self.upload_to = upload_to

def get_col_spec(self, **kwargs):
return self.__class__.__name__.lower()

def bind_processor(self, fp):
def process(value):
root_folder = os.path.dirname(os.path.dirname(__name__))
media_folder = os.path.join(root_folder, 'media', self.upload_to)
os.makedirs(media_folder, exist_ok=True)
name = os.path.join(media_folder, value.filename)
self.storage.save(name, value.stream)
return os.path.join(self.upload_to, value.filename)
return process

def result_processor(self, fp, coltype):
def process(value):
print(f'result: {value}')
return os.path.join(self.upload_to, value.filename)

return process

models.py

from apps import db
from custom_fields.myfield import MyField

class TestModel(db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
email = db.Column(db.String(120), unique=True, nullable=False)
custom_field = db.Column(MyField(upload_to='cindy/he'))

def __repr__(self):
return '<User %r>' % self.username

如果使用 flask-migrate来迁移的话,要注意: 创建带数据库和迁移脚本的项目

Comments:

captcha