Python实现读取TXT文件数据并存进内置数据库SQLite3的方法

Python  /  管理员 发布于 7年前   364

本文实例讲述了Python实现读取TXT文件数据并存进内置数据库SQLite3的方法。分享给大家供大家参考，具体如下：

当TXT文件太大，计算机内存不够时，我们可以选择按行读取TXT文件，并将其存储进Python内置轻量级splite数据库,这样可以加快数据的读取速度，当我们需要重复读取数据时，这样的速度加快所带来的时间节省是非常可观的，比如，当我们在训练数据时，要迭代10万次，即要从文件中读取10万次，即使每次只加快0.1秒，那么也能节省几个小时的时间了。

#创建数据库并把txt文件的数据存进数据库import sqlite3      #导入sqlite3cx = sqlite3.connect('./train.db')  #创建数据库，如果数据库已经存在，则链接数据库；如果数据库不存在，则先创建数据库，再链接该数据库。cu = cx.cursor()           #定义一个游标，以便获得查询对象。cu.execute('create table if not exists train4 (id integer primary key,name text)')  #创建表fr = open('data_sample.txt')    #打开要读取的txt文件i = 0for line in fr.readlines():    #将数据按行插入数据库的表train4中。  cu.execute('insert into train4 values(?,?)',(i,line))  i +=1cu.close()   #关闭游标cx.commit()   #事务提交cx.close()   #关闭数据库

查询数据：

cu.execute('select * from train4 where id = ?',(i,)) #i代表你要读取表train4中某一行的数据result = cu.fetchall()

注：如果前面已经关闭了数据库，那么在查询时要重新打开数据库，并创建游标。这一点要注意一下。

完整的查询程序是这样的：

import sqlite3cx = sqlite3.connect('./train.db')cu = cx.cursor()for i in range(5):  cu.execute('select * from train4 where id = ?',(i,))  result = cu.fetchall()  cx.commit()cu.close()cx.close()

另：这里再为大家附带一个SQLite3数据操作类供大家参考使用：

import sqlite3# ***************************************************# *# * Description: Python操作SQLite3数据库辅助类(查询构造器)# * Author: wangye# *# ***************************************************def _wrap_value(value):  return repr(value)def _wrap_values(values):  return list(map(_wrap_value, values))def _wrap_fields(fields):  for key,value in fields.items():    fields[key] = _wrap_value(value)  return fieldsdef _concat_keys(keys):  return "[" + "],[".join(keys) + "]"def _concat_values(values):  return ",".join(values)def _concat_fields(fields, operator = (None, ",")):  if operator:    unit_operator, group_operator = operator  # fields = _wrap_fields(fields)  compiled = []  for key,value in fields.items():    compiled.append("[" + key + "]")    if unit_operator:      compiled.append(unit_operator)      compiled.append(value)    compiled.append(group_operator)  compiled.pop() # pop last group_operator  return " ".join(compiled)class DataCondition(object):  """    本类用于操作SQL构造器辅助类的条件语句部分    例如:    DataCondition(("=", "AND"), id = 26)    DataCondition(("=", "AND"), True, id = 26)  """  def __init__(self, operator = ("=", "AND"), ingroup = True, **kwargs):    """      构造方法      参数:        operator 操作符，分为(表达式操作符, 条件运算符)        ingroup 是否分组，如果分组，将以括号包含        kwargs  键值元组，包含数据库表的列名以及值 注意这里的等于号不等于实际生成SQL语句符号 实际符号是由operator[0]控制的      例如:      DataCondition(("=", "AND"), id = 26)      (id=26)      DataCondition((">", "OR"), id = 26, age = 35)      (id>26 OR age>35)      DataCondition(("LIKE", "OR"), False, name = "John", company = "Google")      name LIKE 'John' OR company LIKE "Google"    """    self.ingroup = ingroup    self.fields = kwargs    self.operator = operator  def __unicode__(self):    self.fields = _wrap_fields(self.fields)    result = _concat_fields(self.fields, self.operator)    if self.ingroup:      return "(" + result + ")"    return result  def __str__(self):    return self.__unicode__()  def toString(self):    return self.__unicode__()class DataHelper(object):  """    SQLite3 数据查询辅助类  """  def __init__(self, filename):    """      构造方法      参数: filename 为SQLite3 数据库文件名    """    self.file_name = filename  def open(self):    """      打开数据库并设置游标    """    self.connection = sqlite3.connect(self.file_name)    self.cursor = self.connection.cursor()    return self  def close(self):    """      关闭数据库，注意若不显式调用此方法，      在类被回收时也会尝试调用    """    if hasattr(self, "connection") and self.connection:      self.connection.close()  def __del__(self):    """      析构方法，做一些清理工作    """    self.close()  def commit(self):    """      提交事务      SELECT语句不需要此操作，默认的execute方法的      commit_at_once设为True会隐式调用此方法，      否则就需要显示调用本方法。    """    self.connection.commit()  def execute(self, sql = None, commit_at_once = True):    """      执行SQL语句      参数:        sql 要执行的SQL语句，若为None，则调用构造器生成的SQL语句。        commit_at_once 是否立即提交事务，如果不立即提交，        对于非查询操作，则需要调用commit显式提交。    """    if not sql:      sql = self.sql    self.cursor.execute(sql)    if commit_at_once:      self.commit()  def fetchone(self, sql = None):    """      取一条记录    """    self.execute(sql, False)    return self.cursor.fetchone()  def fetchall(self, sql = None):    """      取所有记录    """    self.execute(sql, False)    return self.cursor.fetchall()  def __concat_keys(self, keys):    return _concat_keys(keys)  def __concat_values(self, values):    return _concat_values(values)  def table(self, *args):    """      设置查询的表，多个表名用逗号分隔    """    self.tables = args    self.tables_snippet = self.__concat_keys(self.tables)    return self  def __wrap_value(self, value):    return _wrap_value(value)  def __wrap_values(self, values):    return _wrap_values(values)  def __wrap_fields(self, fields):    return _wrap_fields(fields)  def __where(self):    # self.condition_snippet    if hasattr(self, "condition_snippet"):      self.where_snippet = " WHERE " + self.condition_snippet  def __select(self):    template = "SELECT %(keys)s FROM %(tables)s"    body_snippet_fields = {      "tables" : self.tables_snippet,      "keys" : self.__concat_keys(self.body_keys),     }    self.sql = template % body_snippet_fields  def __insert(self):    template = "INSERT INTO %(tables)s (%(keys)s) VALUES (%(values)s)"    body_snippet_fields = {      "tables" : self.tables_snippet,      "keys" : self.__concat_keys(list(self.body_fields.keys())),      "values" : self.__concat_values(list(self.body_fields.values()))    }    self.sql = template % body_snippet_fields  def __update(self):    template = "UPDATE %(tables)s SET %(fields)s"    body_snippet_fields = {      "tables" : self.tables_snippet,      "fields" : _concat_fields(self.body_fields, ("=",","))    }    self.sql = template % body_snippet_fields  def __delete(self):    template = "DELETE FROM %(tables)s"    body_snippet_fields = {      "tables" : self.tables_snippet    }    self.sql = template % body_snippet_fields  def __build(self):    {      "SELECT": self.__select,      "INSERT": self.__insert,      "UPDATE": self.__update,      "DELETE": self.__delete    }[self.current_token]()  def __unicode__(self):    return self.sql  def __str__(self):    return self.__unicode__()  def select(self, *args):    self.current_token = "SELECT"    self.body_keys = args    self.__build()    return self  def insert(self, **kwargs):    self.current_token = "INSERT"    self.body_fields = self.__wrap_fields(kwargs)    self.__build()    return self  def update(self, **kwargs):    self.current_token = "UPDATE"    self.body_fields = self.__wrap_fields(kwargs)    self.__build()    return self  def delete(self, *conditions):    self.current_token = "DELETE"    self.__build()    #if *conditions:    self.where(*conditions)    return self  def where(self, *conditions):    conditions = list(map(str, conditions))    self.condition_snippet = " AND ".join(conditions)    self.__where()    if hasattr(self, "where_snippet"):      self.sql += self.where_snippet    return self

更多关于Python相关内容感兴趣的读者可查看本站专题：《Python操作SQLite数据库技巧总结》、《Python常见数据库操作技巧汇总》、《Python数据结构与算法教程》、《Python函数使用技巧总结》、《Python字符串操作技巧汇总》、《Python入门与进阶经典教程》及《Python文件与目录操作技巧汇总》

希望本文所述对大家Python程序设计有所帮助。