专栏首页Python进阶之路Python编程 模拟SQL语句 实现对员工信息的增删改查

Python编程 模拟SQL语句 实现对员工信息的增删改查

一、问题描述

用 Python 模拟 sql 语句,实现对员工信息的增删改查。

  • 封装函数,传入参数:文件路径和 sql 命令。
  • 模拟 sql 语句实现对员工信息的现增删改查,并打印结果。

二、Python编程

导入需要的依赖库

# -*- coding: UTF-8 -*-
"""
@Author   :叶庭云
@file     :实训第二次作业
@function :封装函数  根据输入的文件路径和sql命令
            模拟sql语句实现对员工信息的现增删改查
"""
import re
import os

函数式编程

def sql_parse(sql_, key_list):
    """
    解析sql命令字符串,按照key_lis列表里的元素分割sql得到字典形式的命令sql_dic
    :param sql_:
    :param key_list:
    :return:
    """
    sql_list = []
    sql_dic = {}
    for i in key_list:
        b = [j.strip() for j in sql_.split(i)]
        if len(b) > 1:
            if len(sql_.split('limit')) > 1:
                sql_dic['limit'] = sql_.split('limit')[-1]
            if i == 'where' or i == 'values':
                sql_dic[i] = b[-1]
            if sql_list:
                sql_dic[sql_list[-1]] = b[0]
            sql_list.append(i)
            sql_ = b[-1]
        else:
            sql_ = b[0]
        if sql_dic.get('select'):
            if not sql_dic.get('from') and not sql_dic.get('where'):
                sql_dic['from'] = b[-1]
    if sql_dic.get('select'):
        sql_dic['select'] = sql_dic.get('select').split(',')
    if sql_dic.get('where'):
        sql_dic['where'] = where_parse(sql_dic.get('where'))
    return sql_dic


def where_parse(where):
    """
    格式化where字符串为列表where_list,用'and', 'or', 'not'分割字符串
    :param where:
    :return:
    """
    casual_l = [where]
    logic_key = ['and', 'or', 'not']
    for j in logic_key:
        for i in casual_l:
            if i not in logic_key:
                if len(i.split(j)) > 1:
                    ele = i.split(j)
                    index = casual_l.index(i)
                    casual_l.pop(index)
                    casual_l.insert(index, ele[0])
                    casual_l.insert(index + 1, j)
                    casual_l.insert(index + 2, ele[1])
                    casual_l = [k for k in casual_l if k]
    where_list = three_parse(casual_l, logic_key)
    return where_list


def three_parse(casual_l, logic_key):
    """
    处理临时列表casual_l中具体的条件,'staff_id>5'-->['staff_id','>','5']
    :param casual_l:
    :param logic_key:
    :return:
    """
    where_list = []
    for i in casual_l:
        if i not in logic_key:
            b = i.split('like')
            if len(b) > 1:
                b.insert(1, 'like')
                where_list.append(b)
            else:
                key = ['<', '=', '>']
                new_lis = []
                opt = ''
                lis = [j for j in re.split('([=<>])', i) if j]
                for k in lis:
                    if k in key:
                        opt += k
                    else:
                        new_lis.append(k)
                new_lis.insert(1, opt)
                where_list.append(new_lis)
        else:
            where_list.append(i)
    return where_list


def sql_action(sql_dic, titles):
    """
    把解析好的sql_dic分发给相应函数执行处理
    :param sql_dic:
    :param titles:
    :return:
    """
    key = {'select': select,
           'insert': insert,
           'delete': delete,
           'update': update}
    res = []
    for i in sql_dic:
        if i in key:
            res = key[i](sql_dic, titles)
    return res


def select(sql_dic, title_):
    """
    处理select语句命令
    :param sql_dic:
    :param title_:
    :return:
    """
    with open(data_path, 'r', encoding='utf-8') as fh:
        filter_res = where_action(fh, sql_dic.get('where'), title_)
        limit_res = limit_action(filter_res, sql_dic.get('limit'))
        search_res = search_action(limit_res, sql_dic.get('select'), title_)
    return search_res


def insert(sql_dic):
    """
    处理insert语句命令
    :param sql_dic:
    :return:
    """
    with open(data_path, 'r+', encoding='utf-8') as fp:
        data_ = fp.readlines()
        phone_list = [i.strip().split(',')[4] for i in data_]
        ins_count = 0
        if not data_:
            new_id = 1
        else:
            last = data_[-1]
            last_id = int(last.split(',')[0])
            new_id = last_id + 1
        record = sql_dic.get('values').split('/')
        for i in record:
            if i.split(',')[2] in phone_list:
                print('\033[1;31m%s 手机号已存在\033[0m' % i)
            else:
                new_record = '\n%s,%s\n' % (str(new_id), i)
                fp.write(new_record)
                new_id += 1
                ins_count += 1
        fp.flush()  # 刷新记录
    return ['insert successfully!'], [str(ins_count)]


def delete(sql_dic, item):
    """
    处理delete语句命令
    :param sql_dic:
    :param item:
    :return:
    """
    with open(data_path, 'r', encoding='utf-8') as r_file, \
            open('staff_data_bak.txt', 'w', encoding='utf-8') as w_file:
        del_count = 0
        # delete from staff_data.txt where staff_id>=5 and staff_id<=10
        for line in r_file.read().split("\n")[1:]:
            # print(line)
            dic = dict(zip(item.split(','), line.split(',')))
            # print(dic)
            # delete from staff_data.txt where staff_id>=5 and staff_id<=10
            print(dic)
            print(sql_dic.get("where"))
            # exp_1, opt, exp_2 = sql_dic.get("where")
            record_list = []
            try:
                for item in sql_dic.get("where"):
                    print(item)
                    if type(item) is list:
                        exp_1, opt, exp_2 = item
                        print(exp_1, opt, exp_2)
                        flag = eval(f'{dic["staff_id"]}{opt}{exp_2}')
                        print(flag)
                        record_list.append(flag)
                print(all(record_list), record_list)
                if all(record_list):  # 删除满足条件的
                    del_count += 1
                    pass
                else:
                    w_file.write(line + "\n")
            except Exception as e:
                print(e.args[0])
                pass
        w_file.flush()
    # os.remove('staff_data.txt')
    # os.rename('staff_data_bak.txt', 'staff_data.txt')
    return ['delete successfully!'], [str(del_count)]


def update(sql_dic, title_):
    """
    处理update语句命令
    :param sql_dic:
    :param title_:
    :return:
    """
    set_l = sql_dic.get('set').strip().split(',')
    set_list = [i.split('=') for i in set_l]
    update_count = 0
    with open(data_path, 'r', encoding='utf-8') as r_file, \
            open('staff_data_bak.txt', 'w', encoding='utf-8') as w_file:
        for line in r_file:
            dic = dict(zip(title_.split(','), line.strip().split(',')))
            filter_res = logic_action(dic, sql_dic.get('where'))
            if filter_res:
                for i in set_list:
                    k = i[0]
                    v = i[-1]
                    dic[k] = v
                line = [dic[i] for i in title_.split(',')]
                update_count += 1
                line = ','.join(line) + '\n'
            w_file.write(line)
        w_file.flush()
    os.remove('staff_data.txt')
    os.rename('staff_data_bak.txt', 'staff_data.txt')
    return ['update successfully!'], [str(update_count)]


def where_action(fh, where_list, title_):
    """
    具体处理where_list里的所有条件
    :param fh:
    :param where_list:
    :param title_:
    :return:
    """
    res = []
    if len(where_list) != 0:
        for line in fh:
            dic = dict(zip(title_.split(','), line.strip().split(',')))
            if dic['name'] != 'name':
                logic_res = logic_action(dic, where_list)
                if logic_res:
                    res.append(line.strip().split(','))
    else:
        res = [i.split(',') for i in fh.readlines()]
    return res


def logic_action(dic, where_list):
    """
    判断数据文件中每一条是否符合where_list条件
    :param dic:
    :param where_list:
    :return:
    """
    logic = []
    # print(where_list)
    # print(dic)
    for exp in where_list:
        # print(exp)
        if type(exp) is list:
            exp_k, opt, exp_v = exp
            # print(exp_k, opt, exp_v)
            if exp[1] == '=':
                opt = '=='
            logical_char = "'%s'%s'%s'" % (dic[exp_k], opt, exp_v)
            # print(logical_char)
            if opt != 'like':
                exp = str(eval(logical_char))
            else:
                if exp_v in dic[exp_k]:
                    exp = 'True'
                else:
                    exp = 'False'
        logic.append(exp)
    res = eval(' '.join(logic))
    return res


def limit_action(filter_res, limit_l):
    """
    用列表切分处理显示符合条件的数量
    :param filter_res:
    :param limit_l:
    :return:
    """
    if limit_l:
        index = int(limit_l[0])
        res = filter_res[:index]
    else:
        res = filter_res
    return res


def search_action(limit_res, select_list, title_):
    """
    处理需要查询并显示的title和相应数据
    :param limit_res:
    :param select_list:
    :param title_:
    :return:
    """
    res = []
    fields_list_ = title_.split(',')
    if select_list[0] == '*':
        res = limit_res
    else:
        fields_list_ = select_list
        for data_ in limit_res:
            dic = dict(zip(title_.split(','), data_))
            r_l = []
            for i in fields_list_:
                r_l.append((dic[i].strip()))
            res.append(r_l)
    return fields_list_, res

主函数调用

if __name__ == '__main__':
    # 指令关键词列表
    key_lis = ['select', 'insert', 'delete', 'update', 'from', 'into', 'set', 'values', 'where', 'limit']
    while True:
        sql = input('请输入sql命令,退出请输入exit->>>').strip()
        sql = re.sub(' ', '', sql)
        if len(sql) == 0:
            continue
        if sql == 'exit':
            break
        sql_dict = sql_parse(sql, key_lis)
        try:
            data_path = sql_dict['from']
        except KeyError:
            data_path = sql_dict['into']
        with open(data_path, 'r', encoding='utf-8') as f:
            title = f.readline().strip()
        fields_list, fields_data = sql_action(sql_dict, title)
        print('\033[0;32m结果如下:\033[0m')
        print('-'.join(fields_list))
        for data in fields_data:
            print('-'.join(data))

三、测试结果

# 测试sql命令如下:
# select * from staff_data.txt where dept=人事
# select name,age from staff_data.txt where age > 27
# select * from staff_data.txt where enroll_date like 2017
# insert into staff_data.txt values 叶庭云,21,13198497869,算法,2018-7-12
# update into staff_data.txt set dept=算法,phone=13566677787 where staff_id=8
# delete from staff_data.txt where staff_id>=5 and staff_id<=10
# delete from staff_data.txt where staff_id>=5 and staff_id<=10

所有测试结果如下:

本文参与 腾讯云自媒体分享计划 ,欢迎热爱写作的你一起参与!
本文分享自作者个人站点/博客:https://blog.csdn.net/fyfugoyfa复制
如有侵权,请联系 yunjia_community@tencent.com 删除。
登录 后参与评论
0 条评论

相关文章

  • Python MySQL数据库交互

    可能默认的源安装第三库会有点慢,可以配置一下其他的镜像源。Pip安装第三方库网速慢(解决方案)

    忆想不到的晖
  • Python MySQL数据库交互

    可能默认的源安装第三库会有点慢,可以配置一下其他的镜像源。Pip安装第三方库网速慢(解决方案)

    忆想不到的晖
  • 33.企业级开发进阶6:数据库操作

    python作为一个编程语言,在开发B/S或者C/S结构的软件时,不可避免的会设计到和数据库之间的交互操作,和其他高级的面向对象的语言一样,Python在操作数...

    大牧莫邪
  • MySQL与Python的交互

    1、交互类型 1、安装引入模块 安装mysql模块,在windows和ubuntu中 ? windows里安装mysql模块 ? Linux里安装mysql模块...

    用户1332428
  • MySQL与Python的交互1.交互类型2.增删改查(CRUD)3.封装

    Python攻城狮
  • Python MySQL数据库交互

    可能默认的源安装第三库会有点慢,可以配置一下其他的镜像源。Pip安装第三方库网速慢(解决方案)[1]

    ZackSock
  • Python入门进阶教程-数据库操作

    MySQL 是最流行的关系型数据库管理系统。本小节通过 Python 对 MySQL 数据库进行增删改查操作,后期高阶可以通过结合 DataFrame 对文件实...

    小一不二三
  • 数据库(七)

    视图是由一张表或多张表的查询结果构成的一张虚拟表,建立一张视图后会在数据库中保留一个以 frm 后缀结尾的文件,只保留了数据结果,所有的数据都来自 sql 语句...

    py3study
  • 专栏:008:MySQLdb及其银行模拟转账

    谢伟
  • day26.MySQL【Python教程】

    Java帮帮
  • Python小白的数据库入门

    sqlite3 以上两个是主流的关系型数据库,我们观察之后发现,它们与我们熟知的Excel好像也没有什么不同。关系型数据库里面放的都是一张张的表,就如同Exce...

    arcticfox
  • 软件测试所需要掌握的技能

    1.在测试中最重要的文档,他是测试工作的核心,是一组在测试时输入输出的标准,是软件需求的具体对照。编写测试用例,是测试人员的基本功,真正能写好的人并不多。

    用户7880705
  • Django之ORM基础

    ORM简介 ORM概念 对象关系映射(Object Relational Mapping,简称ORM)模式是一种为了解决面向对象与关系数据库存在的互不匹配的现象...

    人生不如戏
  • 如何用Python自动操作数据库?

    我在使用 Python 之前,做数据分析工作的流程,一般是先打开数据库客户端,然后运行一段写好的 SQL 语句,把数据查询出来,然后再把数据复制到 Excel ...

    数据森麟
  • 数据库系统概论考试心得+复习指南

    试后感悟:老师给的大纲完全就是按照试卷的顺序来的,比如触发器是第一题,数据库恢复是第二题。

    wsuo
  • python 操作MySQL数据库

    输出:连接成功: <pymysql.connections.Connection object at 0x00000205AC8E96D0>

    Michael阿明
  • MySQL数据库基础教程(一)-简介

    1970 年 IBM 的 E.F. Codd 博士发表了论文《A Relational Model of Data for Large Shared Data ...

    JavaEdge
  • 「Python爬虫系列讲解」六、Python 数据库知识

    前文回顾: 「Python爬虫系列讲解」一、网络数据爬取概述 「Python爬虫系列讲解」二、Python知识初学 「Python爬虫系列讲解」三、正则表达...

    荣仔_最靓的仔

扫码关注云+社区

领取腾讯云代金券