Python處理SSH和Redis的示例代碼詳解

更新時間：2026年01月23日 09:25:14 作者：半路_出家ren

這篇文章主要為大家詳細(xì)介紹了Python中處理SSH和Redis的相關(guān)方法,文中的示例代碼講解詳細(xì),感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下

一、Python處理SSH

基于Python代碼直接可以遠(yuǎn)程操作Linux服務(wù)器（此類技術(shù)的運(yùn)用：利用Python完成遠(yuǎn)程應(yīng)用系統(tǒng)部署，遠(yuǎn)程監(jiān)控，文件傳輸，無代理模式）

安裝第三方庫：Paramiko

pip install paramiko

Python代碼實(shí)現(xiàn)

第一階段：最基礎(chǔ)的SSH連接

import paramiko

# 第一步：創(chuàng)建SSH客戶端
ssh_client = paramiko.SSHClient()

# 第二步：設(shè)置自動接受未知主機(jī)密鑰（學(xué)習(xí)時使用，生產(chǎn)環(huán)境不建議）
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

# 第三步：連接到服務(wù)器
ssh_client.connect(
    hostname='192.168.102.131',
    port=22,           # SSH默認(rèn)端口
    username='yt',
    password='123.com'
)

print("連接成功！")

# 第四步：執(zhí)行一個簡單命令
# stdin,stdout,stderr = ssh_client.exec_command('pwd')
# stdin,stdout,stderr = ssh_client.exec_command('ls -la')
# stdin,stdout,stderr = ssh_client.exec_command('cd /usr/ && pwd')
cmd = '''
cd /usr
ls -la
pwd
whoami
'''
stdin,stdout,stderr = ssh_client.exec_command(cmd)

# 第五步：讀取命令輸出
output = stdout.read().decode('utf-8')
print(f"命令輸出結(jié)果：{output}")

# 第六步：關(guān)閉連接
ssh_client.close()
print("連接已關(guān)閉")

第二階段：處理命令輸出和錯誤

import paramiko

# 創(chuàng)建一個執(zhí)行命令的函數(shù)
def execute_ssh_command(host, user, password, command):
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    try:
        ssh.connect(hostname=host, username=user, password=password)

        # 執(zhí)行命令
        stdin, stdout, stderr = ssh.exec_command(command)

        # 等待命令完成，并獲取退出狀態(tài)碼
        exit_status = stdout.channel.recv_exit_status()

        # 讀取輸出
        output = stdout.read().decode('utf-8')
        error_output = stderr.read().decode('utf-8')

        print(f"命令: {command}")
        print(f"退出狀態(tài)碼: {exit_status} (0表示成功)")

        if output:
            print("輸出結(jié)果:")
            print(output)

        if error_output:
            print("錯誤信息:")
            print(error_output)

    finally:
        ssh.close()


# 測試不同的命令
commands = ['pwd', 'ls /nonexistent', 'date']

for cmd in commands:
    execute_ssh_command('192.168.102.131', 'yt', '123.com', cmd)
    print("-" * 50)

第三階段：傳輸文件

import paramiko

def upload_file():
    """上傳文件到服務(wù)器"""
    # 先建立連接
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect('服務(wù)器IP', username='用戶名', password='密碼')
    
    # 打開SFTP會話
    sftp = ssh.open_sftp()
    
    # 上傳本地文件到遠(yuǎn)程服務(wù)器
    local_file = 'test.txt'        # 本地文件
    remote_file = '/tmp/test.txt'  # 遠(yuǎn)程路徑
    
    sftp.put(local_file, remote_file)
    print(f"文件上傳成功：{local_file} -> {remote_file}")
    
    # 關(guān)閉連接
    sftp.close()
    ssh.close()

def download_file():
    """從服務(wù)器下載文件"""
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect('服務(wù)器IP', username='用戶名', password='密碼')
    
    sftp = ssh.open_sftp()
    
    # 下載文件
    remote_file = '/var/log/syslog'  # 遠(yuǎn)程文件
    local_file = 'syslog_copy.log'   # 保存到本地
    
    sftp.get(remote_file, local_file)
    print(f"文件下載成功：{remote_file} -> {local_file}")
    
    sftp.close()
    ssh.close()

def list_directory():
    """列出遠(yuǎn)程目錄內(nèi)容"""
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect('服務(wù)器IP', username='用戶名', password='密碼')
    
    sftp = ssh.open_sftp()
    
    # 列出目錄內(nèi)容
    remote_dir = '/tmp'
    files = sftp.listdir(remote_dir)
    
    print(f"目錄 {remote_dir} 中的內(nèi)容：")
    for file in files:
        print(f"  - {file}")
    
    sftp.close()
    ssh.close()

第四階段：封裝成實(shí)用工具

import paramiko

class SimpleSSHClient:
    def __init__(self, host, username, password=None, key_path=None):
        self.host = host
        self.username = username
        self.password = password
        self.key_path = key_path
        self.client = None
    
    def connect(self):
        """建立連接"""
        self.client = paramiko.SSHClient()
        self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        
        if self.key_path:
            # 使用密鑰認(rèn)證
            key = paramiko.RSAKey.from_private_key_file(self.key_path)
            self.client.connect(
                hostname=self.host,
                username=self.username,
                pkey=key
            )
        else:
            # 使用密碼認(rèn)證
            self.client.connect(
                hostname=self.host,
                username=self.username,
                password=self.password
            )
        print(f"已連接到 {self.host}")
    
    def run_command(self, command):
        """執(zhí)行命令并返回結(jié)果"""
        if not self.client:
            print("錯誤：請先調(diào)用connect()方法連接")
            return None
        
        stdin, stdout, stderr = self.client.exec_command(command)
        
        # 獲取結(jié)果
        exit_code = stdout.channel.recv_exit_status()
        output = stdout.read().decode('utf-8')
        error = stderr.read().decode('utf-8')
        
        return {
            'command': command,
            'exit_code': exit_code,
            'output': output,
            'error': error
        }
    
    def disconnect(self):
        """斷開連接"""
        if self.client:
            self.client.close()
            print("連接已關(guān)閉")
    
    def upload(self, local_path, remote_path):
        """上傳文件"""
        sftp = self.client.open_sftp()
        sftp.put(local_path, remote_path)
        sftp.close()
        print(f"上傳完成：{local_path} -> {remote_path}")
    
    def download(self, remote_path, local_path):
        """下載文件"""
        sftp = self.client.open_sftp()
        sftp.get(remote_path, local_path)
        sftp.close()
        print(f"下載完成：{remote_path} -> {local_path}")

# 使用示例
def main():
    # 創(chuàng)建客戶端（使用密碼）
    ssh = SimpleSSHClient(
        host='服務(wù)器IP',
        username='用戶名',
        password='密碼'
    )
    
    # 或者使用密鑰
    # ssh = SimpleSSHClient(
    #     host='服務(wù)器IP',
    #     username='用戶名',
    #     key_path='/path/to/private_key'
    # )
    
    # 連接
    ssh.connect()
    
    # 執(zhí)行命令
    result = ssh.run_command('ls -la')
    print(f"命令: {result['command']}")
    print(f"退出碼: {result['exit_code']}")
    print(f"輸出:\n{result['output']}")
    
    # 上傳文件
    ssh.upload('本地文件.txt', '/tmp/遠(yuǎn)程文件.txt')
    
    # 斷開連接
    ssh.disconnect()

if __name__ == "__main__":
    main()

二、Python處理Redis

1.驗(yàn)證安裝

import redis
# 測試連接
try:
    r = redis.Redis(host='192.168.102.136', password='123.com',port=6379, db=0)
    r.ping()
    print("? Redis 連接成功！")
except Exception as e:
    print(f"? 連接失敗: {e}")

2. 連接 Redis

import redis

# 基本連接
r = redis.Redis(
    host='192.168.102.136',  # 主機(jī)
    port=6379,         # 端口
    db=0,              # 數(shù)據(jù)庫編號
    password='123.com',     # 密碼（如果有）
    decode_responses=True  # 自動解碼為字符串
)

# 連接池方式（推薦）
pool = redis.ConnectionPool(
    host='localhost',
    port=6379,
    db=0,
    max_connections=10  # 最大連接數(shù)
)
r = redis.Redis(connection_pool=pool)

3.基礎(chǔ)操作

import redis

# 連接池方式（推薦）
pool = redis.ConnectionPool(
    host='192.168.102.136',
    password='123.com',
    port=6379,
    db=0,
    max_connections=10  # 最大連接數(shù)
)
r = redis.Redis(connection_pool=pool)

# 設(shè)置鍵值對
r.set('name', 'Alice')
r.set('age', '25')
r.setex('temp_key', 60, '臨時數(shù)據(jù)')  # 60秒后過期

# 獲取值
name = r.get('name')
print(f"姓名: {name}")  # Alice

# 批量處理
r.mset({'key1':'value1','key2':'value2'})
values = r.mget(['key1','key2'])
print(f"批量獲取:{values}")

# 自增/自減
r.set('counter', 0)
r.incr('counter')  # 增加到1
r.incrby('counter', 5)  # 增加到6
r.decr('counter')  # 減少到5
print(r.get('counter'))

print("="*20)
# 列表操作

# 添加元素
r.lpush('mylist', 'first')   # 左側(cè)插入
r.rpush('mylist', 'second')  # 右側(cè)插入
r.lpush('mylist', 'zero')

# 獲取元素
print(r.lrange('mylist', 0, -1))  # 獲取所有
print(r.lindex('mylist', 1))      # 獲取索引1的元素

# 彈出元素
left_item = r.lpop('mylist')
right_item = r.rpop('mylist')
print(r.lrange('mylist', 0, -1))  # 獲取所有

# 列表長度
length = r.llen('mylist')
print(length)

print("="*20)

# 設(shè)置哈希字段
r.hset('user:1001', 'name', 'Bob')
r.hset('user:1001', 'age', 30)
r.hset('user:1001', 'city', 'Beijing')

# 批量設(shè)置
r.hmset('user:1002', {'name': 'Charlie', 'age': 25})

# 獲取字段
name = r.hget('user:1001', 'name')
print(name)
all_fields = r.hgetall('user:1001')
print(f"所有字段: {all_fields}")

# 刪除字段
r.hdel('user:1001', 'city')
all_fields = r.hgetall('user:1001')
print(f"所有字段: {all_fields}")

# 集合操作
# 添加元素
r.sadd('tags', 'python', 'redis', 'database')
r.sadd('tags', 'python')  # 重復(fù)元素不會添加

# 獲取所有元素
tags = r.smembers('tags')
print(f"標(biāo)簽: Python
SSH
Redis
")

# 集合運(yùn)算
r.sadd('set1', 'a', 'b', 'c')
r.sadd('set2', 'b', 'c', 'd')

# 交集
inter = r.sinter('set1', 'set2')
print(f"交集: {inter}")

# 并集
union = r.sunion('set1', 'set2')
print(f"并集: {union}")

# 差集
diff = r.sdiff('set1', 'set2')
print(f"差集: {diff}")

# 有序集合
# 添加帶分?jǐn)?shù)的元素
r.zadd('scores', {'Alice': 90, 'Bob': 85, 'Charlie': 95})

# 獲取排名
rank = r.zrank('scores', 'Bob')  # 升序排名
rev_rank = r.zrevrank('scores', 'Bob')  # 降序排名

# 獲取分?jǐn)?shù)范圍
top3 = r.zrevrange('scores', 0, 2, withscores=True)
print(f"前三名: {top3}")

# 獲取分?jǐn)?shù)
score = r.zscore('scores', 'Alice')

官方文檔：https://redis.io/documentation

Python Redis 文檔：https://redis-py.readthedocs.io/

到此這篇關(guān)于Python處理SSH和Redis的示例代碼詳解的文章就介紹到這了,更多相關(guān)Python處理SSH和Redis內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: