使用orm—sqlalchemy

  1 from sqlalchemy.orm import sessionmaker
  2 from sqlalchemy import create_engine
  3 from sqlalchemy.ext.declarative import declarative_base
  4 from sqlalchemy import Column, Integer, String, DateTime, Boolean
  5 
  6 engine = create_engine('mysql://root:@localhost:3306/news?charset=utf8')
  7 Base = declarative_base()
  8 Session = sessionmaker(bind=engine)
  9 
 10 
 11 class News(Base):
 12     __tablename__ = 'news'
 13     id = Column(Integer, primary_key=True)
 14     title = Column(String(200), nullable=False)
 15     content = Column(String(20000), nullable=False)
 16     created_at = Column(DateTime)
 17     types = Column(String(10), nullable=False)
 18     author = Column(String(20))
 19     view_count = Column(Integer)
 20     is_valid = Column(Boolean)
 21 
 22 
 23 class OrmText(object):
 24     def __init__(self):
 25         self.session = Session()
 26 
 27     def add_one(self):
 28         """新增一条记录"""
 29         try:
 30             orm_obj = News(
 31                 title='伊朗感染新冠病毒人数达9000人',
 32                 content='新闻内容',
 33                 types='百家',
 34             )
 35             self.session.add(orm_obj)
 36             self.session.commit()
 37             return orm_obj
 38         except:
 39             print('Error')
 40             self.session.rollback()
 41 
 42     def get_one(self):
 43         """获取一条数据"""
 44         return self.session.query(News).get(6)
 45 
 46     def get_more(self):
 47         """获取多条数据"""
 48         return self.session.query(News).filter_by(is_valid=True)
 49 
 50     def get_all(self):
 51         """获取所有数据"""
 52         list = []
 53         for sert in self.session.query(News).all():
 54             list.append(sert)
 55         return list
 56 
 57     def update_data(self, update_id):
 58         """修改数据"""
 59         obj = self.session.query(News).get(update_id)
 60         if obj:
 61             obj.title = '全球市场进入暴跌模式 至少11国发生“熔断”'
 62             self.session.add(obj)
 63             self.session.commit()
 64             return True
 65         else:
 66             return False
 67 
 68     def delete_data(self, delete_id):
 69         """删除数据"""
 70         data = self.session.query(News).get(delete_id)
 71         self.session.delete(data)
 72         self.session.commit()
 73 
 74 # 建表语句
 75 News.metadata.create_all(engine)
 76 
 77 # 添加数据
 78 obj = OrmText()
 79 rest = obj.add_one()
 80 print(rest.id)
 81 
 82 # 获取一条记录
 83 rest = obj.get_one()
 84 if rest:
 85     print('%s:%s' % (rest.id, rest.title))
 86 else:
 87     print("数据不存在")
 88 
 89 # 获取多条数据
 90 obj = OrmText()
 91 rest = obj.get_more()
 92 print(rest.count())
 93 print(rest)
 94 for i in rest:
 95     print('%s:%s' % (i.id, i.title))
 96 
 97 # 获取所有数据
 98 obj = OrmText()
 99 for i in obj.get_all():
100     print(i.id, i.title)
101 
102 # 修改数据
103 obj = OrmText()
104 text = obj.update_data(3)
105 print(text)
106 
107 # 删除数据
108 data = OrmText()
109 data.delete_data(7)