SQLAlchemy

時間  2019-11-10
標籤 sqlalchemy 简体版
原文   原文鏈接

一. pymysql

import pymysql


conn = pymysql.connect(host="10.37.129.3",port=3306,user="egon",passwd="123456",db="homework",charset="utf8")

cursor = conn.cursor()


#方式一:

sql = "select * from course where cid=1"
effect_row = cursor.execute(sql)


# 方式二：
sql = "select * from course where cid='%s'" %(1,)
effect_row = cursor.execute(sql)


# 方式三  普通  列表
sql = "select * from course where cid='%s'"

effect_row = cursor.execute(sql,1)
effect_row = cursor.execute(sql,[1])


# 方式四  字典格式

sql = "select * from course where cid='%(u)s'"
effect_row = cursor.execute(sql,{"u":1})


row_1 = cursor.fetchone()

cursor.close()
conn.close()

print(row_1)
基本用法 
import pymysql


conn = pymysql.connect(host="10.37.129.3",port=3306,user="egon",passwd="123456",db="student_info",charset="utf8")

cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)



sql = "select * from student_info where sid > %s "

effect_row = cursor.execute(sql,[1])


row_1 = cursor.fetchone()
row_2 = cursor.fetchone()
row_3 = cursor.fetchone()

cursor.close()
conn.close()

print(row_1)
print(row_2)
print(row_3)
字典格式顯示 
import pymysql

conn = pymysql.connect(host="localhost",user='root',password='',database="db666")
cursor = conn.cursor()
sql = "insert into userinfo(username,password) values('asdfasdf','123123')"
cursor.execute(sql)
conn.commit()
print(cursor.lastrowid)
cursor.close()
conn.close()
獲取自增ID cursor.lastrowid 
import pymysql


conn = pymysql.connect(host="10.37.129.3",port=3306,user="egon",passwd="123456",db="student_info",charset="utf8")

cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)


#插入一行
# sql = "insert into student_info(sname,gender,class_id) VALUES('alex1','女',2)"

#插入多行
sql = "insert into student_info(sname,gender,class_id) VALUES('alex1','女',2),('alex2','女',2),('alex3','女',2)"

r = cursor.execute(sql)

#或

sql = "insert into userinfo(username,password) values(%s,%s)"
# 受影響的行數
r = cursor.executemany(sql,[('egon','sb'),('laoyao','BS')])




conn.commit()
cursor.close()
conn.close()

print(r)

插入 演示

插入演示
插入演示

二. 做業要求：html

做業：
        參考表結構：
            用戶類型

            用戶信息

            權限

            用戶類型&權限
        功能：

            # 登錄、註冊、找回密碼
            # 用戶管理
            # 用戶類型
            # 權限管理
            # 分配權限

        特別的：程序僅一個可執行文件
做業要求 
create database wuSir default character set utf8 collate utf8_general_ci;

use wuSir;


create table auth_info(
    aid int not null auto_increment primary key,
    auth_name varchar(32),
    unique(auth_name)
    )engine=innodb default charset=utf8;


create table user_info(
    uid int not null auto_increment primary key,
    name varchar(32),
    passwd varchar(32),
    sex ENUM("男","女")
    )engine=innodb default charset=utf8;

create table user_auth(
    id int,
    auth_id int,
    constraint auth_info foreign key(auth_id) references auth_info(aid),
    constraint user_info foreign key(id) references user_info(uid)
    )engine=innodb default charset=utf8;



insert into auth_info(auth_name) values("訂單管理"),("用戶管理"),("菜單管理"),("權限分配"),("Bug管理");

insert into user_info(name,passwd,sex) values("alex",123,"男"),("egon",123,"男");

insert into user_auth(id,auth_id) values(1,1),(2,1),(2,2),(2,3);
建立庫表 SQL 
import pymysql


user = input("please input name: ").strip()
passwd = input("please input passwd: ").strip()


conn = pymysql.connect(host="10.37.129.3",port=3306,user="egon",passwd="123456",db="wuSir",charset="utf8")

cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)

sql = "select * from user_info where name=%s and passwd=%s"

cursor.execute(sql,(user,passwd,))
result = cursor.fetchone()

if result["name"] == user and result["passwd"] == passwd:
    uid = result["uid"]
    # print(uid)
    sql = "select auth_name from auth_info where aid in (select auth_id from user_auth where id =%s)"

    cursor.execute(sql,(uid,))
    result = cursor.fetchall()
    for i in result:
        print(i["auth_name"])

else:
    print("error")

cursor.close()
conn.close()
Python 代碼

二. SQLAlchemy

SQLAlchemy是Python編程語言下的一款ORM框架，該框架創建在數據庫API之上，使用關係對象映射進行數據庫操做，簡言之即是：將對象轉換成SQL，而後使用數據API執行SQL並獲取執行結果。python

pip3 install SQLAlchemy

SQLAlchemy自己沒法操做數據庫，其必須以來pymsql等第三方插件，Dialect用於和數據API進行交流，根據配置文件的不一樣調用不一樣的數據庫API，從而實現對數據庫的操做，如：mysql

MySQL-Python
    mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
   
pymysql
    mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
   
MySQL-Connector
    mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
   
cx_Oracle
    oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
   
更多詳見：http://docs.sqlalchemy.org/en/latest/dialects/index.html

a. 內部處理(執行SQL語句)sql

使用 Engine/ConnectionPooling/Dialect 進行數據庫操做，Engine使用ConnectionPooling鏈接數據庫，而後再經過Dialect執行SQL語句。數據庫

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy import create_engine
  
  
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5)
  
# 執行SQL
# cur = engine.execute(
#     "INSERT INTO hosts (host, color_id) VALUES ('1.1.1.22', 3)"
# )
  
# 新插入行自增ID
# cur.lastrowid
  
# 執行SQL
# cur = engine.execute(
#     "INSERT INTO hosts (host, color_id) VALUES(%s, %s)",[('1.1.1.22', 3),('1.1.1.221', 3),]
# )
  
  
# 執行SQL
# cur = engine.execute(
#     "INSERT INTO hosts (host, color_id) VALUES (%(host)s, %(color_id)s)",
#     host='1.1.1.99', color_id=3
# )
  
# 執行SQL
# cur = engine.execute('select * from hosts')
# 獲取第一行數據
# cur.fetchone()
# 獲取第n行數據
# cur.fetchmany(3)
# 獲取全部數據
# cur.fetchall()
執行Sql語句 演示

b. ORM功能使用編程

使用 ORM/Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect 全部組件對數據進行操做。根據類建立對象，對象轉換成SQL，執行SQL。session

1. 建立表oracle

#!/usr/bin/env python
# -*- coding:utf-8 -*-
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
 
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5)
 
Base = declarative_base()
 
# 建立單表
class Users(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    extra = Column(String(16))
 
    __table_args__ = (
    UniqueConstraint('id', 'name', name='uix_id_name'),
        Index('ix_id_name', 'name', 'extra'),
    )
 
 
# 一對多
class Favor(Base):
    __tablename__ = 'favor'
    nid = Column(Integer, primary_key=True)
    caption = Column(String(50), default='red', unique=True)
 
 
class Person(Base):
    __tablename__ = 'person'
    nid = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)
    favor_id = Column(Integer, ForeignKey("favor.nid"))
 
 
# 多對多
class Group(Base):
    __tablename__ = 'group'
    id = Column(Integer, primary_key=True)
    name = Column(String(64), unique=True, nullable=False)
    port = Column(Integer, default=22)
 
 
class Server(Base):
    __tablename__ = 'server'
 
    id = Column(Integer, primary_key=True, autoincrement=True)
    hostname = Column(String(64), unique=True, nullable=False)
 
 
class ServerToGroup(Base):
    __tablename__ = 'servertogroup'
    nid = Column(Integer, primary_key=True, autoincrement=True)
    server_id = Column(Integer, ForeignKey('server.id'))
    group_id = Column(Integer, ForeignKey('group.id'))
 
 
def init_db():
    Base.metadata.create_all(engine)
 
 
def drop_db():
    Base.metadata.drop_all(engine)
建立表 知識點 
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

Base = declarative_base()


# 建立單表
class UserType(Base):
    __tablename__ = 'usertype'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(32), nullable=True, index=True)

class Users(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True, autoincrement=True)  #主鍵    自增
    name = Column(String(32), nullable=True)                    #不爲空
    email = Column(String(16), unique=True)                     #惟一
    user_type_id = Column(Integer,ForeignKey("usertype.id"))    #外鍵

    __table_args__ = (
        UniqueConstraint('name', 'email', name='uix_id_name'),  #聯合惟一索引
        Index('ix_n_ex','name', 'email',),
    )


engine = create_engine("mysql+pymysql://egon:123456@10.37.129.3:3306/day63?charset=utf8", max_overflow=5)


#建立表
def create_db():
    Base.metadata.create_all(engine)

#刪除表
def drop_db():
    Base.metadata.drop_all(engine)
建立表 演示 
insert into usertype(title) values("普通用戶"),("白金用戶"),("黑金用戶");

insert into users(name,email,user_type_id) values("alex","163",1),("egon","173",1),("wuSir","183",2);


select * from usertype;
    +----+--------------+
    | id | title        |
    +----+--------------+
    |  1 | 普通用戶     |
    |  2 | 白金用戶     |
    |  3 | 黑金用戶     |
    +----+--------------+


select * from users;
    +----+-------+-------+--------------+
    | id | name  | email | user_type_id |
    +----+-------+-------+--------------+
    |  1 | alex  | 163   |            1 |
    |  2 | egon  | 173   |            1 |
    |  3 | wuSir | 183   |            2 |
    +----+-------+-------+--------------+
表插入數據

2. 表操做演示框架

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

Base = declarative_base()


# 建立單表
class UserType(Base):
    __tablename__ = 'usertype'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(32), nullable=True, index=True)

class Users(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True, autoincrement=True)  #主鍵    自增
    name = Column(String(32), nullable=True)                    #不爲空
    email = Column(String(16), unique=True)                     #惟一
    user_type_id = Column(Integer,ForeignKey("usertype.id"))    #外鍵

    __table_args__ = (
        UniqueConstraint('name', 'email', name='uix_id_name'),  #聯合惟一索引
        Index('ix_n_ex','name', 'email',),
    )


engine = create_engine("mysql+pymysql://egon:123456@10.37.129.3:3306/day63?charset=utf8", max_overflow=5)

Session = sessionmaker(bind=engine)
session = Session()


# ret = session.query(Users)
    #SELECT users.id AS users_id, users.name AS users_name, users.email AS users_email, users.user_type_id AS users_user_type_id FROM users

# ret = session.query(Users).all()
    #[<__main__.Users object at 0x1037620f0>, <__main__.Users object at 0x103762160>, <__main__.Users object at 0x1037621d0>]

# ret = session.query(Users.name, Users.email).all()
#     [('alex', '163'), ('egon', '173'), ('wuSir', '183')]

# ret = session.query(Users).filter_by(name='alex').all()
#     [<__main__.Users object at 0x103759198>]

# ret = session.query(Users).filter_by(name='alex').first()
#     <__main__.Users object at 0x103758240>

#user_list = session.query(UserType.id,UserType.title).filter(UserType.id>=1).all()
#   [(1, '普通用戶'), (2, '黃金用戶')]    

session.commit()
session.close()
查詢 語法 
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

Base = declarative_base()


# 建立單表
class UserType(Base):
    __tablename__ = 'usertype'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(32), nullable=True, index=True)

class Users(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True, autoincrement=True)  #主鍵    自增
    name = Column(String(32), nullable=True)                    #不爲空
    email = Column(String(16), unique=True)                     #惟一
    user_type_id = Column(Integer,ForeignKey("usertype.id"))    #外鍵

    __table_args__ = (
        UniqueConstraint('name', 'email', name='uix_id_name'),  #聯合惟一索引
        Index('ix_n_ex','name', 'email',),
    )


engine = create_engine("mysql+pymysql://egon:123456@10.37.129.3:3306/day63?charset=utf8", max_overflow=5)


Session = sessionmaker(bind=engine)
session = Session()


#方式一：
# obj = UserType(title="普通用戶")
# obj1 = UserType(title="黃金用戶")
# session.add(obj)
# session.add(obj1)


#方式二:

objs= [
  Users(name="alex",email="163",user_type_id=1),
  Users(name="egon",email="173",user_type_id=1),
  Users(name="tom",email="183",user_type_id=2)
]

session.add_all(objs)


session.commit()
session.close()
插入語法 
#刪除
# session.query(Users).filter(Users.id>1).delete()


#修改

#session.query(Users).filter(Users.id > 2).update({"name" : "099"})
#session.query(Users).filter(Users.id > 2).update({Users.name: Users.name + "099"}, synchronize_session=False)
#session.query(Users).filter(Users.id > 2).update({"num": Users.num + 1}, synchronize_session="evaluate")
刪除 修改 語法

3. 基本操做編程語言

from sqlalchemy import create_engine,String,Integer,Column
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

db_name = "mysql+mysqlconnector://root:123456@localhost:3306/test"

Base = declarative_base()

class User(Base):
    __tablename__ = 'user'
    id = Column(Integer,primary_key=True)
    name = Column(String(50))

engine = create_engine(db_name)
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()


#交互式

    #增長
        方式一:
            u = User()
            u.id = 1
            u.name = "Lily"
            session.add(u)    #會話層保存
            session.commit()

        方式二:
            u2 = User(id=2,name="Bob")
            session.add(u2)
            session.commit()

    #查詢

        # filter_by
            u2 = session.query(User).filter_by(id=2)                #一個列表
            u3 = session.query(User).filter_by(id=2).first()        #取出第一個值    是一個對象
            u3.id        
            u3.name

        # filter
            u = session.query(User).filter(User.id==3).first()        #輸出對象
        
        # all
            session.query(User).all()                                #輸出列表    裏面是對象

        # order_by
            u1 = session.query(User).order_by('id').all()        #依照id的順序排序
            u1[0].id
            u1[0].id

    #刪除
        session.delete(u3)
        session.commit()

        session.query(User).filter_by(id=3).delete()
基本操做
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程