大兵是
一个想全能的运维

python使用mysql/redis/mongo

python 操作 mysql

python安装mysql依赖

python3.5版本以上 本文是python3.6

pip install mysql-python

##### 装不上了下载地址
https://pypi.python.org/pypi/mysqlclient   linux
http://www.lfd.uci.edu/~gohlke/pythonlibs/#mysqlclient    windows

mac 安装mysqlclient

brew install mysql-connector-c
cd /usr/local/Cellar/mysql-connector-c/6.1.11/bin/mysql_config

# 修改前先备份
cp  mysql_config mysql_config.backup

# 使用vi修改配置文件
sudo vi mysql_config

# 114 gg跳转到 114行

将
> # Create options 
> libs="-L$pkglibdir"
> libs="$libs -l "

替换为

> # Create options 
> libs="-L$pkglibdir"
> libs="$libs -lmysqlclient -lssl -lcrypto"

检测mysql依赖是否安装成功

进入虚拟环境 
>> import MySQLdb 

使用mysql连接数据库

https://mysqlclient.readthedocs.io

连接数据库配置

import MySQLdb

# 获取连接
try:
    conn = MySQLdb.connect(
        host="47.93.33.159",
        port=3306,
        user='dabing',
        password='Xxhxtdgs123',
        db='news',
        charset='utf8'
    )
        # 获取数据
    cursor = conn.cursor();
    cursor.execute('SELECT * from `news`;')
    result = cursor.fetchone()
    print(result)

    # 关闭连接
    conn.close()

except MySQLdb.Error as e:
    print('Error: %s' %e)

python操作redis

安装redis

pip install redis

# 使用
import redis 
r = redis.StrictRedis(host='localhost',port=6379,db=0,password='')
r.set('foo','bar')
r.get('foo')

示例

# Author dabing
# date: '2018/1/9 15:22'
# -*- coding:utf-8 -*-

import redis
# redis.StrictRedis(host='192.168.44.11', port=6379, password='dabing')

class TestString(object):

    def __init__(self):
        self.r = redis.StrictRedis(host='192.168.44.11',port=6379,password='dabing')

    def test_set(self):
        rest = self.r.set('user1','Jonh')
        print(rest)

    def get_set(self):
        return self.r.get('girl')

    def test_mset(self):
        rest = self.r.mset(user3={'name':'dabing','age':79})
        return rest

    def test_mget(self):
        return self.r.mget('user3','user1')

    def test_del(self):
        return self.r.delete('user3')


class TestList(object):

    def __init__(self):
        self.r = redis.StrictRedis(host='192.168.44.11', port=6379, password='dabing')

    def test_push(self):
        # 插入
        t = ['Amy','Jonh']
        self.r.lpush('l_eat',*t)
        # 查询
        rest = self.r.lrange('l_eat',0,-1)
        return  rest

    def test_pop(self):
        rest = self.r.lpop('l_eat')
        return rest


class TestSet(object):
    """
    sadd/srem -- 添加/删除元素
    sismember -- 判断是否为set的一个元素
    smembers -- 返回改set的所有成员
    sdiff -- 返回集合之间的差异
    sinter -- 返回集合的交集
    sunion -- 返回集合的并集
    """

    def __init__(self):
        self.r = redis.StrictRedis(host='192.168.44.11',port=6379,password='dabing')


    def test_sadd(self):

        self.r.sadd('class',*['dabing','zhangsan','wangwu','dabing','xiaoyu'])

class TestHash(object):
    """
    hset/hget -- 设置/获取散列值
    hmset/hmget -- 设置/获取多对散列值
    hsetnx -- 如果散列存在, 则不设置
    hkeys/hvals -- 返回所有的key/value
    hlen -- 返回散列包含的域(field)的数量
    hdel -- 删除散列的域(field)
    hexists -- 判断是否存在
    """
    def __init__(self):
        self.r = redis.StrictRedis(host='192.168.44.11',port=6379,password='dabing')


def main():
    # string 操作
    str_obj = TestString()
    # str_obj.test_set()
    # print(str_obj.get_set())
    # print(str_obj.test_mset())
    # print(str_obj.test_mget())
    # print(str_obj.test_del())


    # list 操作
    list_obj = TestList()
    # print(list_obj.test_push())
    # print(list_obj.test_pop())


    # 集合操作
    set_obj = TestSet()
    set_obj.test_sadd()




if __name__ == '__main__':
    main()

python操作mongo

安装

pip install pymongo

连接mongo数据库

# 方式1
client = MongoClient()

# 方式2
client = MongoClient('localhost',27017)

# 方式3
client = MongoClient('mongodb://localhost:27017')

示例

# Author dabing
# date: '2018/1/7 18:19'
# -*- coding:utf-8 -*-
from datetime import datetime

from pymongo import MongoClient
from bson.objectid import ObjectId


class TestMongo(object):

    def __init__(self):
        self.client = MongoClient('192.168.44.11',27017)
        self.db = self.client['blog']


    def add_one(self):
        """ 增加一条 """
        post = {
            'title': "new title11111111111111",
            'content': "new 222222222222222222222",
            'created_at': datetime.now()
        }

        return self.db.blog.posts.insert_one(post)


    def get_one(self):
        """ 查询一条数据 """
        return self.db.blog.posts.find_one()


    def get_more(self):
        """ 查询多条数据 """
        return self.db.blog.posts.find()


    def get_from_iod(self,oid):
        """ 根据记录的oid来获取数据 """

        return self.db.blog.posts.find_one({'_id':ObjectId(oid)})


    def update(self):
        """ 修改数据 """
        # 修改一条数据
        return self.db.blog.posts.update_one({"x":10},{'$inc':{'x':30}})
        # 修改多条数据
        return self.db.blog.posts.update_many({"x": 10}, {'$inc': {'x': 30}})

    def delete(self):
        """ 删除数据 """
        # 删除一条
        return self.db.blog.posts.delete_one({'x':40})
        # 删除 多条
        return self.db.blog.posts.delete_many({'x': 40})


def main():
    obj = TestMongo()
    # rest = obj.add_one()
    # print(rest)
    # print(rest.inserted_id)

    # print("###############查询一条##################")
    # print(obj.get_one())
    # print("###############查询多条##################")
    # for rest in obj.get_more():
    #     print(rest)
    # print("###############查询为id的条##################")
    # print(obj.get_from_iod('5a52c90d28efcc840c5ed290'))

    # print(obj.update().modified_count)

    print(obj.delete().deleted_count)


if __name__ == '__main__':
    main()
打赏

未经允许不得转载: » python使用mysql/redis/mongo

分享到:更多 ()

评论 1

  • 昵称 (必填)
  1. #1

    厉害厉害~

    大131312个月前 (06-06)回复