Python SqlAlchemy使用方法

    Python SqlAlchemy使用方法html

 2.初始化鏈接:python

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
import os
basedir = os.path.abspath(os.path.dirname(__file__))

# 建立對象的基類:
Base = declarative_base()

 

 2.建立表:sql

# 建立單表
class Users(Base):
# 表名
__tablename__ = 'users'
# 表的結構:
id = Column(Integer, primary_key=True)
username = Column(String(32))
password = Column(String(16))

 

 初始化數據庫:數據庫

 def __init__(self, username, password):
# 初始化數據庫
self.username = username
self.password = password

建立聯合惟一索引session

# __table_args__(
# UniqueConstraint("列名1","列名2","聯合惟一索引名"),
# index("索引名","列名1","列名2"),
# ) #建立聯合惟一索引
__table_args__ = (
UniqueConstraint('id', 'username', name='uix_id_name'),
Index('ix_id_name', 'username', 'password'),
)

 3.建立映射post

# 建立映射
class UserORMHelper(object):
def __init__(self, database_name):
DATABASE_PATH = os.path.join(basedir, database_name)
SQLALCHEMY_DATABASE_URI = 'sqlite:///' + DATABASE_PATH
self.engine = create_engine(SQLALCHEMY_DATABASE_URI)

 建立表測試

def create_db(self):
# 建立表
Base.metadata.create_all(self.engine) # 建立表
Session = sessionmaker(bind=self.engine)
self.session = Session()

 刪除表ui

def drop_db(self):
# 刪除表
Base.metadata.drop_all(self.engine) #刪除表

 4.接下來就是簡單的針對數據庫的增刪改查操做了,這些操做我是爲了方便測試安裝本身的作了些修改,後面我會附上測試的代碼lua

   插入數據url

 

def addUser(self, users):
# 插入數據
isSucess=False
usrsList=self.query_all_with_user_name_password(users)
if(usrsList and len(usrsList)>0):#用戶已經註冊
return False
self.session.add(users)
self.session.commit()
isSucess=True
return isSucess

 

刪除數據
def delete(self,users):
# 刪除數據
isSucess=False
user1 = self.session.query(Users).filter_by(username=users.username).first()
self.session.delete(user1)
self.session.commit()
isSucess=True
return isSucess

 修改

(1)匹配username修改password

def update_user_extra_by_user_name(self, users):
# 匹配而且修改password
isSucess = False
self.session.query(Users).filter(Users.username == users.username).update({"password": users.password}, synchronize_session='evaluate')
self.session.commit()
isSucess=True
return isSucess

 (2)匹配password修改username

def update_user_name_by_user_extra(self, users):
# 匹配而且修改name
isSucess = False
self.session.query(Users).filter(Users.password == users.password).update({"username": users.username}, synchronize_session='evaluate')
self.session.commit()
isSucess=True
return isSucess

 

查詢

安password查詢全部

def query_all_with_user_extra(self, users):
# 查詢相同password數據
userList = self.session.query(Users).filter_by(password=users.password).all()
return userList

 安username查詢全部,可是隻取其一

def query_one(self,users):
# 查詢相同name數據,取一個
userList=self.session.query(Users).filter_by(username=users.username).limit(1).all()
self.session.commit()
return userList

  安username查詢可是隻取第一條數據

    def query_first_with_user_name(self, users):
        # 查詢相同name數據,取第一個
        userList=self.session.query(Users).filter_by(username=users.username).first()
        return userList

 安username,password查詢全部

def query_all_with_user_name_password(self, users):
# 查詢全部相同於name,password數據
userList=self.session.query(Users).filter_by(username=users.username,password=users.password).all()
return userList

 安username查詢全部

def query_all_with_user_name(self, users):
# 查詢全部相同name數據
userList=self.session.query(Users).filter_by(username=users.username).all()
return userList

 下面的是數據庫的測試代碼,寫測試我用的是TDD

import unittest
from UserORM import Users,UserORMHelper




TEST_DB = 'test.db'

class UserORMHelperTestCase(unittest.TestCase):



    def setUp(self):
        """Set up a blank temp database before each test"""
        self.helper=UserORMHelper("user.db")
        self.helper.drop_db()
        self.helper.create_db()


    def tearDown(self):
        """Destroy blank temp database after each test"""
        self.helper.drop_db()

    def test_add(self):
        # 插入測試
        isSuccess=self.helper.addUser(Users("tome", "tom jobn"))
        self.assertTrue(isSuccess)
        isSuccess=self.helper.addUser(Users("tome", "tom jobn"))
        self.assertFalse(isSuccess)


    def test_delete(self):
        # 刪除測試
        isSuccess = self.helper.addUser(Users("tome", "tom jobn"))
        self.assertTrue(isSuccess)
        isSucess=self.helper.delete(Users("tome", "tom jobn"))
        self.assertTrue(isSucess)

    def test_query_all(self):
        # 查詢全部相同name數據測試
         isSuccess=self.helper.addUser(Users("tome", "tom jobn"))
         self.assertTrue(isSuccess)
         isSuccess=self.helper.addUser(Users("tome", "tom jobn2"))
         self.assertTrue(isSuccess)
         isSuccess=self.helper.addUser(Users("tome", "tom jobn3"))
         self.assertTrue(isSuccess)
         userList=self.helper.query_all_with_user_name(Users("tome", "tom jobn"))
         # print userList
         self.assertEqual(len(userList),3)


    def test_query_none(self):
        # 查詢相同name數據,取第一個  測試
         userList=self.helper.query_first_with_user_name(Users("tom", "tom job"))
         self.assertEqual(userList,None)

    def test_query_one(self):
        # 查詢相同name數據,取一個   測試
        isSuccess = self.helper.addUser(Users("tome", "tom jobn"))
        self.assertTrue(isSuccess)
        isSuccess = self.helper.addUser(Users("tome", "tom jobn2"))
        self.assertTrue(isSuccess)
        isSuccess = self.helper.addUser(Users("tome", "tom jobn3"))
        self.assertTrue(isSuccess)
        userList = self.helper.query_one(Users("tome", "tom jobn4"))
        self.assertEqual(len(userList), 1)


    def test_revise_extra(self):
        isSuccess = self.helper.addUser(Users("tome", "tom jobn"))
        self.assertTrue(isSuccess)

        isSuccess=self.helper.update_user_extra_by_user_name(Users("tome", "tomUpdate"))

        self.assertTrue(isSuccess)
        userList = self.helper.query_all_with_user_name(Users("tome", "tom job"))
        self.assertEqual(len(userList), 1)


    def test_recise_name(self):
        isSuccess = self.helper.addUser(Users("tome", "tom jobn"))
        self.assertTrue(isSuccess)
        isSuccess = self.helper.update_user_name_by_user_extra(Users("tomeUpdate", "tom jobn"))
        self.assertTrue(isSuccess)
        userList = self.helper.query_all_with_user_name(Users("tomeUpdate", "tom job"))
        self.assertEqual(len(userList),1)















if __name__ == '__main__':
    unittest.main()
相關文章
相關標籤/搜索