操作MySQL

1)Windows中安裝python和pycharm
2)ubuntu中安裝python和pycharm

安裝步驟不做贅述，pycharm運(yùn)行腳本

#!/usr/bin/env python import MySQLdb #get connection try:con = MySQLdb.connect(host='localhost',user='root',passwd='12346',port=3308,db='sakila',charset='utf8') except MySQLdb.Error as e:print('error:%s'% e) cursor = con.cursor() cursor.execute('SELECT * FROM `store`') rest = cursor.fetchone() print(rest) #close connection con.close()

3）查詢數(shù)據(jù)庫

#!/usr/bin/env python import MySQLdb class MysqlQuery(object):def __init__(self):self.get_conn()def get_conn(self):# get connectiontry:self.conn = MySQLdb.connect(host='localhost',user='root',passwd='123456',port=3308,db='sakila',charset='utf8')except MySQLdb.Error as e:print('error:%s' % e)cursor = self.conn.cursor()cursor.execute('SELECT * FROM `store`')rest = cursor.fetchone()print(rest)def close_conn(self):try:if self.conn:# close connectionself.conn.close()except MySQLdb.Error as e:print('Error:%s'%e)def get_one(self):#prepare SQL/*雖然定義類型為int，可使用string*/sql='SELECT * FROM `store` where `store_id` =%s'#get cursorcursor=self.conn.cursor()cursor.execute(sql,('1',))print(cursor.rowcount)rest=dict(zip([k[0] for k in cursor.description],cursor.fetchone()))print(rest)print(rest['store_id'])self.close_conn()return rest def main():obj = MysqlQuery()rest = obj.get_one();print(rest['store_id'])if __name__ == '__main__':main()/*取走所有數(shù)據(jù)，形成數(shù)組*/ rest = [dict(zip([k[0] for k in cursor.description],row))for row in cursor.fetchall()]

zip([iterable, …])
Python的一個內(nèi)建函數(shù)，它接受一系列可迭代的對象作為參數(shù)，將對象中對應(yīng)的元素打包成一個個tuple（元組），然后返回由這些tuples組成的list（列表）。若傳入?yún)?shù)的長度不等，則返回list的長度和參數(shù)中長度最短的對象相同。利用*號操作符，可以將list unzip（解壓）。

dict()作用：dict() 函數(shù)用于創(chuàng)建一個字典。返回一個字典。

class dict(**kwarg) class dict(mapping, **kwarg) class dict(iterable, **kwarg) /*kwargs -- 關(guān)鍵字mapping -- 元素的容器。iterable -- 可迭代對象 */

4）更新數(shù)據(jù)

def add_one(self):row_count=0try:sql = ("insert into `film`(`title`,`description`,`language_id`) value" "(%s,%s,%s);")cursor = self.conn.cursor()cursor.execute(sql, ('chia', 'ashajhsjah','1'))self.conn.commit()except:print('error')self.conn.rollback()row_count=cursor.rowcountcursor.close()self.close_conn()return row_count

5）ORM：SQLAlChemy

pip install SQLAlchemyimport sqlalchemy

declarative_base() 創(chuàng)建了一個 BaseModel 類，這個類的子類可以自動與一個表關(guān)聯(lián)。
增刪改查

#!/usr/bin/python #coding=utf-8 from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from sqlalchemy import Column, Integer, String, DateTime, Boolean import datetime #engine=create_engine('數(shù)據(jù)庫類型://用戶名:密碼@ip:端口/數(shù)據(jù)庫名') engine=create_engine('mysql://root:123456@localhost:3308/sakila') Base=declarative_base() Session = sessionmaker(bind=engine) class Store(Base):__tablename__='country'#數(shù)據(jù)庫表名country_id = Column(Integer, primary_key=True)country = Column(String(200), nullable=False)last_update = Column(String(2000), nullable=False) class MysqlOrmTest(object):def __init__(self):self.session=Session()def add_one(self):new_obj=Store(country_id='130',country='hhhsahsa',last_update=datetime.datetime.now()#此處需要import datetime)self.session.add(new_obj)self.session.commit()return new_objdef get_one(self):return self.session.query(Store).get(1)def update_date(self):obj=self.session.query(Store).get(1)obj.manager_staff_id=1self.session.add(obj)self.session.commit()return objdef delete_data(self):data=self.session.query(Store).get(3)self.session.delete(data)self.session.commit() def main():# rest = obj.add_one()# print(dir(rest))# print(obj.get_one().title)# print(obj.get_more().count())# for row in obj.get_more():# print(row.title)# print(obj.update_data())if __name__ == '__main__':main()

6）項(xiàng)目實(shí)戰(zhàn)

使用pycharm專業(yè)版，選擇flask框架，代碼如下：

from flask import Flask app = Flask(__name__) @app.route('/hello') def hello_world():return 'Hello World!hello' if __name__ == '__main__':app.run(debug=True) ##flask支持熱部署

簡單搭建flask架構(gòu)網(wǎng)站
本人使用pycharm開發(fā)flask項(xiàng)目，可以利用工具導(dǎo)入工具包：

##引入相關(guān)包 from flask_sqlalchemy import SQLAlchemy from flask import Flask, render_template, flash, redirect, url_for, abort, requestapp = Flask(__name__) ##配置數(shù)據(jù)庫 app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:123456@localhost:3308/flaskdb' db=SQLAlchemy(app) ##完成ORM映射 class News(db.Model):__tablename__='news'id = db.Column(db.Integer, primary_key=True)title = db.Column(db.String(200), nullable=False)img_url = db.Column(db.String(2000), nullable=False)content = db.Column(db.String(2000), nullable=True)is_valid = db.Column(db.String(2000), nullable=True)created_at = db.Column(db.String(2000), nullable=True)updated_at = db.Column(db.String(2000), nullable=True)news_type = db.Column(db.String(2000), nullable=True)def __repr__(self):return '<News %r>' % self.title ##python生成HTML效果不好 @app.route('/hello') def hello_world():return 'Hello World!hello' ##使用渲染引擎Jinja2 @app.route('/') def index():news_list=News.query.all()return render_template("index.html",news_list=news_list)@app.route('/cat/<name>/') def cat(name):news_list=News.query.filter(News.news_type==name)return render_template('cat.html',new_list=news_list)@app.route('/detail/<int:pk>/') def detail(pk):new_obj=News.query.get(pk)return render_template('detail.html',new_obj=new_obj)@app.route('/admin/') @app.route('/admin/<int:page>/') def admin(page=None):return render_template("admin/index.html")@app.route('/admin/add/', methods=['GET', 'POST']) def add():return render_template("admin/add.html")@app.route('/admin/update/<int:pk>/', methods=['GET', 'POST']) def update(pk):return render_template("admin/update.html")@app.route('/admin/delete/<int:pk>/', methods=['POST']) def delete(pk):return 'no'if __name__ == '__main__':app.run(debug=True)

操作Redis

1) Redis安裝

sudo apt-get update sudo apt-get install redis-server##啟動Redis服務(wù)器 redis-server##查看 redis 是否啟動？ redis-cli

2）Redis命令

Set animal 'cat' get animal##添加value append animal 'dog'mset user1 'chu' user2 'yao' mget user1 user2set num 9incr/decr num /*增加減少1*/set user:chuyao;age:45 'asasasasa'

列表（list）相關(guān)操作

lpush/rpush q1 'chu' 'yao' 'Amy'/*從左、右插入數(shù)據(jù)*/ lrange/*獲取指定長度的數(shù)據(jù)*/ ltrim/*截取一定長度的數(shù)據(jù)*/ lpop/rpop/*移除最左、右的元素并返回*/ lpushx/rpushx --key/* key存在時候才插入數(shù)據(jù)，不存在時不做任何處理*/

集合（Set）相關(guān)操作

sadd/srem /*添加、刪除元素*/ sismember /*判斷是否為set的一個元素*/ smembers /*返回該集合的所有成員*/ sdiff /*返回一個集合與其他集合的差異*/ sinter/*返回幾個集合的交集*/ sunion/*返回幾個集合的并集*/

散列（hash）相關(guān)操作

3）redis-py連接

import redis r=redis.StrictRedis(host='120.95.132.174',port=6379,db=0) user1=r.get('user1') print(user1)

注意，如果是遠(yuǎn)程連接數(shù)據(jù)庫，需要修改Redis配置文件。
1）注釋掉bind 127.0.0.1可以使所有的ip訪問redis。
2）修改辦法：protected-mode no

4）Python 操作String類型

import redisclass TestString(object):def __init__(self):self.r=redis.StrictRedis(host='120.95.132.174',port=6379,db=0)def test_set(self):rest=self.r.set('user2','Amy');print(rest)def test_get(self):rest=self.r.get('user1')print restreturn restdef test_mset(self):d={'user3':'Bob','user4':'BobX'}rest=self.r.mset(d)print(rest)return restdef test_mget(self):l=['user1','user2']rest=self.r.mget(l)print(rest)return restdef test_del(self):rest=self.r.delete('user1')print (rest) def main():str_obj=TestString();# str_obj.test_set();str_obj.test_get();# str_obj.test_mset();# str_obj.test_mget();# str_obj.test_del(); if __name__=='__main__':main()

5）項(xiàng)目實(shí)戰(zhàn)

新聞數(shù)據(jù)，Hash
新聞ID,String
分頁數(shù)據(jù)，List
排序，Sorted Set

總結(jié)

以上是生活随笔為你收集整理的Python操作主流数据库的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。