最新消息:20190717 VPS服务器:Vultr新加坡,WordPress主题:大前端D8,统一介绍入口:关于

[已解决]Flask的SQLAlchemy中给数据库添加关系

Flask crifan 891浏览 0评论

之前已经可以实现基本的数据库的字段了:

定义:

#!/usr/bin/python
# -*- coding: UTF-8 -*-
from sipevents import app
from . import db
# from app import db
# from datetime import datetime
class User(db.Model):
    __tablename__ = ‘wechat_users’
    # Columns
    openid = db.Column(db.String(64), primary_key=True)
    province = db.Column(db.String(16))
    headimgurl = db.Column(db.String(256))
    language = db.Column(db.String(8))
    city = db.Column(db.String(16))
    country = db.Column(db.String(32))
    sex = db.Column(db.SmallInteger)
    nickname = db.Column(db.String(128))
    def __init__(self,
                 openid,
                 province = “”,
                 headimgurl = “”,
                 language = “”,
                 city = “”,
                 country = “”,
                 sex = 0,
                 nickname = “”):
        self.openid     = openid
        self.province   = province
        self.headimgurl = headimgurl
        self.language   = language
        self.city       = city
        self.country    = country
        self.sex        = sex
        self.nickname   = nickname
    def __repr__(self):
        return ‘<User %r %s>’ % (self.nickname, self.openid)
class Event(db.Model):
    __tablename__ = “events”
    # Columns
    id = db.Column(db.Integer, primary_key = True, autoincrement = True)
    user_openid = db.Column(db.String(64))
    title = db.Column(db.String(128))
    start_date = db.Column(db.DateTime)
    end_date = db.Column(db.DateTime)
    location = db.Column(db.String(256))
    user_limit = db.Column(db.Integer)
    is_public = db.Column(db.Boolean)
    description = db.Column(db.Text)
    def __init__(self,
                 id,
                 user_openid,
                 title = “”,
                 start_date = None,
                 end_date = None,
                 user_limit = 0,
                 is_public = False,
                 description = “”):
        self.id = id
        self.user_openid = user_openid
        self.title = title
        self.start_date = start_date
        self.end_date = end_date
        self.user_limit = user_limit
        self.is_public = is_public
        self.description = description
    def __repr__(self):
        return ‘<Event id=%d user_openid=%s title=%s>’ % (self.id, self.user_openid, self.title)

使用:

#!flask/bin/python
# -*- coding: UTF-8 -*-
# from migrate.versioning import api
# from config import SQLALCHEMY_DATABASE_URI
# from config import SQLALCHEMY_MIGRATE_REPO
# import os.path
# db.create_all()
# if not os.path.exists(SQLALCHEMY_MIGRATE_REPO):
#     api.create(SQLALCHEMY_MIGRATE_REPO, ‘database repository’)
#     api.version_control(SQLALCHEMY_DATABASE_URI, SQLALCHEMY_MIGRATE_REPO)
# else:
#     api.version_control(SQLALCHEMY_DATABASE_URI, SQLALCHEMY_MIGRATE_REPO, api.version(SQLALCHEMY_MIGRATE_REPO))
from sipevents import db
from sipevents import app
from sipevents import models
from sipevents import User, Event
from datetime import  datetime, timedelta
db.create_all()
# test User create/query/delete
crifanUserJson = {
  “province”: u”江苏”,
  “openid”: “oswjmv4X0cCXcfkIwjoDfCkeTVVY”,
  “headimgurl”: “http://wx.qlogo.cn/mmopen/ajNVdqHZLLDYtIJicNl7MjwZK5c1lxAJZ253c9v3JzDib7GeE5OFrWiaRqsK1ruW1HmGaziaYETV5vQhIIbic6wHKFQ/0”,
  “language”: u”zh_CN”,
  “city”: u”苏州”,
  “country”: u”中国”,
  “sex”: 1,
  “privilege”: [],
  “nickname”: u”礼貌”
}
app.logger.debug(‘crifanUserJson=%s’, crifanUserJson)
existedUser = User.query.filter_by(openid=crifanUserJson[‘openid’]).first()
app.logger.debug(‘existedUser=%s’, existedUser)
if existedUser :
    # has exsited this user
    # callbackOkStr = u’已存在此用户 existedUser=%r’ % (existedUser)
    app.logger.debug(u’已存在此用户 openid=%s, nickname=%s’, crifanUserJson[‘openid’], crifanUserJson[‘nickname’])
else :
    crifanUser = User(openid = crifanUserJson[‘openid’],
                      province = crifanUserJson[‘province’],
                      headimgurl = crifanUserJson[‘headimgurl’],
                      language = crifanUserJson[‘language’],
                      city = crifanUserJson[‘city’],
                      country = crifanUserJson[‘country’],
                      sex = crifanUserJson[‘sex’],
                      nickname = crifanUserJson[‘nickname’])
    app.logger.debug(‘crifanUser=%s’, crifanUser)
    db.session.add(crifanUser)
db.session.commit()
    app.logger.debug(‘added crifanUser=%s’, crifanUser)
currentUsers = User.query.all()
app.logger.debug(‘currentUsers=%s’, currentUsers)
currentUsersCount = len(currentUsers)
app.logger.debug(‘currentUsersCount=%s’, currentUsersCount)
if currentUsersCount > 0  :
    firstUser = currentUsers[0]
    app.logger.debug(‘firstUser=%s’, firstUser)
    db.session.delete(firstUser)
db.session.commit()
    app.logger.debug(‘deleted firstUser=%s’, firstUser)
# test Event create/query/delete
startDatetime = datetime.now()
app.logger.debug(‘startDatetime=%s’, startDatetime)
oneDayDelta = timedelta(days=1)
app.logger.debug(‘oneDayDelta=%s’, oneDayDelta)
endDatetime = startDatetime + oneDayDelta
app.logger.debug(‘endDatetime=%s’, endDatetime)
eventDict = {
    ‘id’            : 1,
    ‘user_openid’   : crifanUserJson[‘openid’],
    ‘title’         : u”讨论日历的公众号名字”,
    ‘start_date’    : startDatetime,
    ‘end_date’      : endDatetime,
    ‘user_limit’    : 5,
    ‘is_public’     : False,
    ‘description’   : u”头脑风暴讨论公众号的名字”,
}
app.logger.debug(‘eventDict=%s’, eventDict)
existedEvent= Event.query.filter_by(id=eventDict[‘id’]).first()
app.logger.debug(‘existedEvent=%s’, existedEvent)
if existedEvent :
    # has exsited this event
    app.logger.debug(‘已存在此活动 existedEvent=%r’, existedEvent)
else :
    eventObj = Event(id = eventDict[‘id’],
                    user_openid = eventDict[‘user_openid’],
                    title = eventDict[‘title’],
                    start_date = eventDict[‘start_date’],
                    end_date = eventDict[‘end_date’],
                    user_limit = eventDict[‘user_limit’],
                    is_public = eventDict[‘is_public’],
                    description = eventDict[‘description’])
    app.logger.debug(‘eventObj=%s’, eventObj)
    db.session.add(eventObj)
db.session.commit()
    app.logger.debug(‘added eventObj=%s’, eventObj)
currentEvents= Event.query.all()
app.logger.debug(‘currentEvents=%s’, currentEvents)
currentEventsCount = len(currentEvents)
app.logger.debug(‘currentEventsCount=%s’, currentEventsCount)
if currentEventsCount > 0  :
    firstEvent = currentEvents[0]
    app.logger.debug(‘firstEvent=%s’, firstEvent)
    db.session.delete(firstEvent)
db.session.commit()
    app.logger.debug(‘deleted firstEvent=%s’, firstEvent)

现在需要去添加两者的关系

参考:

Declaring Models — Flask-SQLAlchemy Documentation (2.1)

Column and Data Types — SQLAlchemy 1.1 Documentation

Quickstart — Flask-SQLAlchemy Documentation (2.1)

快速入门 — Flask-SQLAlchemy 2.0 documentation

转载请注明:在路上 » [已解决]Flask的SQLAlchemy中给数据库添加关系

发表我的评论
取消评论

表情

Hi,您需要填写昵称和邮箱!

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址
46 queries in 0.105 seconds, using 18.71MB memory