租戶隔離
一個(gè)客戶就是一個(gè)租戶爷肝，每個(gè)租戶的數(shù)據(jù)在數(shù)據(jù)表中都有個(gè)一個(gè)tenantid字段用來(lái)與其他租戶隔離
- 租戶識(shí)別
from flask import g, request
def get_tenant_from_request():
auth = validate_auth(request.headers.get('Authorization'))
return Tenant.query.get(auth.tenant_id)
def get_current_tenant():
rv = getattr(g, 'current_tenant', None)
if rv is = None:
rv = get_tenant_from_request()
g.current_tenant = rv
return rv
- 自動(dòng)租戶隔離
例如慷嗜，每個(gè)租戶有自己的Project筐骇，像下面這樣批量直接修改project又忘記帶上tenantid查詢字段录淡，就會(huì)把其他租戶的project一并修改了
def batch_update_projects(ids, changes):
projects = Project.query.filter(
Project.id.in_(ids) &
Project.status != ProjectStatus.INVISIBLE
)
for project in projects:
update_projects(project, changes)
我們可以override Project的query屬性以及使用sqlalchemy的event listener來(lái)自動(dòng)的進(jìn)行租戶隔離，也就是說(shuō)我們?cè)谑褂胦rm查詢的時(shí)候亏娜，會(huì)自動(dòng)帶上tenentid查詢字段厦章，這樣無(wú)論是修改、刪除照藻、查詢都只會(huì)是操作自己租戶下面的數(shù)據(jù)，畢竟在orm中汗侵，刪除和修改都是要先查詢的嘛
# model
class Project(TenantBoundMixin, db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(100))
status = db.Column(db.Integer)
def __repr__(self):
return '<Project name=%r>' % self.name
from sqlalchemy.ext.declarative import declared_attr
?
# 覆寫(xiě)Project類的query_class屬性
class TenantQuery(db.Query):
current_tenant_constrained = True
def tenant_unconstrained_unsafe(self):
rv = self._clone()
rv.current_tenant_constrained = False
return rv
# 通過(guò)mixin的方式幸缕，讓Project類多繼承這個(gè)類
# 從而覆寫(xiě)了Project類的query_class屬性
class TenantBoundMixin(object):
query_class = TenantQuery
# 為project表定義了一個(gè)tenant_id字段
@declared_attr
def tenant_id(cls):
return db.Column(db.Integer, db.ForeignKey('tenant.id'))
?
@declared_attr
def tenant(cls):
return db.relationship(Tenant, uselist=False)
?
# event_listener的意思是，每次執(zhí)行例如Project.query的時(shí)候都會(huì)執(zhí)行
# 下面的鉤子函數(shù)晰韵，
# 這個(gè)鉤子函數(shù)會(huì)給query對(duì)象帶上ilter_by(tenant=get_current_tenant())
@db.event.listens_for(TenantQuery, 'before_compile', retval=True)
def ensure_tenant_constrained(query):
for desc in query.column_descriptions:
if hasattr(desc['type'], 'tenant') and \
query.current_tenant_constrained:
query = query.filter_by(tenant=get_current_tenant())
return query
接下來(lái)我們來(lái)看一下使用
# 只查詢我這個(gè)租戶名下的project42
>>> Project.query.all()
[<Project name='project42'>]
# 去掉租戶約束发乔，查詢所有project信息
>>> Project.query.tenant_unconstrained_unsafe().all()
[<Project name='project1'>, Project.name='project2', ...]
審計(jì)日志
def log(action, message=None):
data = {
'action': action
'timestamp': datetime.utcnow()
}
if message is not None:
data['message'] = message
if request:
data['ip'] = request.remote_addr
user = get_current_user()
if user is not None:
data['user'] = User
db.session.add(LogMessage(**data))
更多詳細(xì)信息，請(qǐng)參考
本篇文章由一文多發(fā)平臺(tái)ArtiPub自動(dòng)發(fā)布
