Python连接mysql[ORM之sqlalchemy]

#!/usr/bin/env python
# -*- coding: utf-8 -*-
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column,Integer,String, DateTime,Boolean,desc,asc
from sqlalchemy.orm import sessionmaker
import datetime
# create_engine()返回一个Engine的实例
engine = create_engine("mysql+pymysql://root:[email protected]:3306/news?charset=utf8",
                       encoding='utf8')
# 通过declarative_base()功能创建一个基类
Base = declarative_base()
#  创建会话，绑定实例
Session = sessionmaker(bind=engine)


class News(Base):
    '''构造一个类，映射表news'''
    __tablename__ = 'news'
    id = Column(Integer, primary_key=True ,autoincrement=True)
    title = Column(String(200), nullable=False)
    content = Column(String(2000), nullable=False)
    created_at = Column(DateTime,nullable=True)
    types = Column(String(10), nullable=False)
    image = Column(String(300), nullable=False)
    author = Column(String(20), nullable=True)
    view_count = Column(Integer,default=0)
    is_valid = Column(Boolean,default=1)


class OrmTest(object):
    '''调用类之前，创建会话'''
    def __init__(self):
        self.session = Session()

    # 新增一条数据
    def add_one(self):
        '''新增一条或多条数据'''
        new_obj = News(
            title = '标题',
            content = '内容',
            types = '百家',
            created_at = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        )
        new_obj2 = News(
            title='标题1',
            content='内容1',
            types='百家1',
            created_at=datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        )
        # 新增一条数据
        self.session.add(new_obj)
        # 新增多条数据，使用[]
        self.session.add_all([new_obj,new_obj2])
        #新增后，提交，或者回滚.rollback()
        self.session.commit()
        return new_obj

    # 查询一条数据,返回一个object
    def get_one(self):
        '''查询一条或多条数据，返回'''
        return self.session.query(News).get(2)

    # 查询多条数据,filter,返回一个object list
    def get_more(self):
        return self.session.query(News).filter(
            News.is_valid==True,News.types=='百家',News.author.is_(None)
        ).order_by(desc(News.id)).all()

    # 查询多条数据,filter_by
    def get_more2(self):
        return self.session.query(News).filter_by(
            is_valid=True,types='百家'
        ).order_by(desc(News.id)).all()

    # 修改一条数据
    def update_date(self, pk):
        '''修改一条数据'''
        obj_list = self.session.query(News).filter(News.id == pk)
        if obj_list:
            for obj in obj_list:
                obj.is_valid = 1
                self.session.add(obj)
            self.session.commit()
            return True
        return False

    # 修改多条数据
    def update_more_date(self,pk):
        '''修改多条数据'''
        data_list = self.session.query(News).filter(News.types==pk)
        if data_list:
            for item in data_list:
                item.is_valid = 1
                self.session.add(item)
            self.session.commit()
            return True
        return False

    # 删除数据
    def delete_date(self, pk):
        '''删除数据'''
        data_list = self.session.query(News).filter(News.id < pk)
        n = 0
        if data_list:
            for item in data_list:
                self.session.delete(item)
                n +=1
            self.session.commit()
            return '删除%s条数据' % n
        return '条件没有数据要删除'

if __name__ == '__main__':
    obj = OrmTest()
    rest=obj.get_one()
    print(rest.title)

查询：

查询一条数据，可以使用get(id),
查询多条数据，filter或者filter_by, 不同的是写法问题，结果是一样，