編輯推薦:

本文來自于博客,本文詳細介紹了如何使用Python操作MySQL、使用Python操作Redis及項目實戰。

操作MySQL

1)Windows中安裝python和pycharm

2)ubuntu中安裝python和pycharm

安裝步驟不做贅述，pycharm運行腳本

#!/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)查詢數據庫

#!/usr/bin/env python

import MySQLdb

class MysqlQuery(object):

def __init__(self):

self.get_conn()

def get_conn(self):

# get connection

try:

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 connection

self.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 cursor

cursor=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()

/*取走所有數據，形成數組*/

rest = [dict(zip([k[0] for k in cursor.description],row))for row in cursor.fetchall()]

zip([iterable, …])

Python的一個內建函數，它接受一系列可迭代的對象作為參數，將對象中對應的元素打包成一個個tuple(元組)，然后返回由這些tuples組成的list(列表)。若傳入參數的長度不等，則返回list的長度和參數中長度最短的對象相同。利用*號操作符，可以將list unzip(解壓)。

dict()作用：dict() 函數用于創建一個字典。返回一個字典。

class dict(**kwarg)

class dict(mapping, **kwarg)

class dict(iterable, **kwarg)

/*

kwargs -- 關鍵字

mapping -- 元素的容器。

iterable -- 可迭代對象

*/

4)更新數據

def add_one(self):

row_count=0

try:

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.rowcount

cursor.close()

self.close_conn()

return row_count

5)ORM：SQLAlChemy

pip install SQLAlchemy

import sqlalchemy

declarative_base() 創建了一個 BaseModel 類，這個類的子類可以自動與一個表關聯。

增刪改查

#!/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('數據庫類型://用戶名:密碼@ip:端口/數據庫名')

engine=create_engine('mysql://root:123456@localhost:3308/sakila')

Base=declarative_base()

Session = sessionmaker(bind=engine)

class Store(Base):

__tablename__='country'#數據庫表名

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_obj

def 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=1

self.session.add(obj)

self.session.commit()

return obj

def 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)項目實戰

使用pycharm專業版，選擇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架構網站

本人使用pycharm開發flask項目，可以利用工具導入工具包：

##引入相關包

from flask_sqlalchemy import SQLAlchemy

from flask import Flask, render_template, flash, redirect, url_for, abort, request

app = Flask(__name__)

##配置數據庫

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 '' % 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//')

def cat(name):

news_list=News.query.filter(News.news_type==name)

return render_template('cat.html',new_list=news_list)

@app.route('/detail//')

def detail(pk):

new_obj=News.query.get(pk)

return render_template('detail.html',new_obj=new_obj)

@app.route('/admin/')

@app.route('/admin//')

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//', methods=['GET', 'POST'])

def update(pk):

return render_template("admin/update.html")

@app.route('/admin/delete//', 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服務器

redis-server

##查看 redis 是否啟動？

redis-cli

2)Redis命令

Set animal 'cat'

get animal

##添加value

append animal 'dog'

mset user1 'chu' user2 'yao'

mget user1 user2

set num 9

incr/decr num /*增加減少1*/

set user:chuyao;age:45 'asasasasa'

列表(list)相關操作

lpush/rpush q1 'chu' 'yao' 'Amy'/*從左、右插入數據*/

lrange/*獲取指定長度的數據*/

ltrim/*截取一定長度的數據*/

lpop/rpop/*移除最左、右的元素并返回*/

lpushx/rpushx --key/* key存在時候才插入數據，不存在時不做任何處理*/

集合(Set)相關操作

sadd/srem /*添加、刪除元素*/

sismember /*判斷是否為set的一個元素*/

smembers /*返回該集合的所有成員*/

sdiff /*返回一個集合與其他集合的差異*/

sinter/*返回幾個集合的交集*/

sunion/*返回幾個集合的并集*/

散列(hash)相關操作

3)redis-py連接

import redis

r=redis.StrictRedis(host='120.95.132.174',port=6379,db=0)

user1=r.get('user1')

print(user1)

注意，如果是遠程連接數據庫，需要修改Redis配置文件。

1)注釋掉bind 127.0.0.1可以使所有的ip訪問redis。

2)修改辦法：protected-mode no

4)Python 操作String類型

import redis

class 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 rest

return rest

def test_mset(self):

d={

'user3':'Bob',

'user4':'BobX'

}

rest=self.r.mset(d)

print(rest)

return rest

def test_mget(self):

l=['user1','user2']

rest=self.r.mget(l)

print(rest)

return rest

def 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)項目實戰

新聞數據，Hash

新聞ID,String

分頁數據，List

排序，Sorted Set

總結

以上是生活随笔為你收集整理的python多线程详解_Python多线程详解-Python-火龙果软件的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。