Created
November 28, 2012 21:30
-
-
Save YurZ/4164707 to your computer and use it in GitHub Desktop.
Basic Python 3 script to read fixed length data and upload into SQLite Database
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python3 | |
## Author: Yuri Zhylyuk <yuri@zhylyuk.com> ## | |
# Amend parameters below for your needs: | |
## *import_file* - file where data gets imported from. | |
## Script is supposed to be ran in the same directory where import_file is located. | |
## *db_file* - name of the SQLite DB file to be created (for the first import) or overwritten. | |
## *dic* - dictionary of field names and their lengths. | |
# Script also writes a basic log file (again in the same directory) | |
######### PARAMETERS Start ######### | |
import_file = "test_file.txt" | |
db_file = "import.sqlite3" | |
dic = { | |
"course_code":10 | |
,"course_post_code":5 | |
,"camp_location":27 | |
,"course_fee_type":1 | |
} | |
########## PARAMETERS End ########## | |
import sqlite3, re, os, logging | |
col_str = "" | |
re_str = "(.{" | |
for name, length in dic.items(): | |
col_str = col_str + ', "' + name + '"' | |
re_str = re_str + str(length) + "})(.{" | |
col_str = col_str.strip(",") | |
re_str = re_str[:-3] | |
logger = logging.getLogger('import_log') | |
hdlr = logging.FileHandler('import_log.log') | |
formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s') | |
hdlr.setFormatter(formatter) | |
logger.addHandler(hdlr) | |
logger.setLevel(logging.INFO) | |
print("\n\n") | |
print("Hi. This is Fix Length 2 SQLite Importer\n\n") | |
print("Working with the following parameters:") | |
print("\t Working directory:", os.getcwd()) | |
print("\t Importing from File:", import_file) | |
print("\t Data will be stored in database:", db_file) | |
logger.info("\t\t--- SCRIPT EXECUTION STARTED ---") | |
logger.info("* Working directory: %s", os.getcwd()) | |
logger.info("* Importing from File: %s", import_file) | |
logger.info("* Database: %s", db_file) | |
print("Starting...\n") | |
conn = sqlite3.connect(db_file) | |
cursor = conn.cursor() | |
cursor.execute("drop table if exists import") | |
conn.commit() | |
cr_import_tbl = "create table import " + "(" + col_str + ")" | |
cursor.execute(cr_import_tbl) | |
conn.commit() | |
print("Table import (re)created") | |
logger.info("Table import (re)created") | |
import_insert = "insert into import values (" + ("?," * len(dic)).strip(",") + ")" | |
print('Reading file', import_file, '...') | |
f = open(import_file, 'r') | |
i = 0 | |
for line in f: | |
row = re.match(re_str,line).groups() | |
cursor.execute(import_insert, row) | |
i = i + 1 | |
f.close() | |
conn.commit() | |
print(i, "rows from", import_file, "uploaded into database") | |
logger.info("%i rows uploaded into database", i) | |
print("Check log file for details") | |
logger.info("\t\t--- SCRIPT EXECUTION FINISHED ---\n") | |
cursor.close() | |
conn.close() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Thanks for this, 5 years later it's a great starting point.