CSVファイルを作る

import csv
from os import path

def create(filepath,headers_list):
    if path.exists(filepath):
        raise FileExistsError("もうある")
    try:
        with open(filepath, "wt", encoding="utf-8", newline="") as file_object:
            writer = csv.writer(file_object)
            writer.writerow(headers_list)
        return filepath
    except Exception as e:
        raise Exception(e)  

ファイルの存在確認をして、なければ作る。
csvモジュールをimportしておいて、csv.witerを使って書き込むと、カンマのエスケープなどを自動でやってくれるのでありがたい。

一行分の情報をリストで渡すと一行分追記する

def append(filepath, line):
    # 1行渡すとCSVに1行追記する 
    try:
        with open(filepath, "a", encoding="utf-8", newline="") as file_object:
            writer = csv.writer(file_object)
            writer.writerow(line)
        return filepath
    except Exception as e:
        raise Exception(e)

一件の処理が終わったら一行分情報を追記していく。
処理の結果をためておいて一気に書き込む方法を採用していると、処理が途中でコケたときに泣く。私は3万件の処理をうっかり一気に回してしまって8時間後に処理がコケてすべてが無になりました。やめよう。
というわけでちゃんと一行一行保存するようにしました。

ついでにCSVを読み込む機能もくっつけてCSV管理クラスまとめ。

class LocalCsvFile():
    def __init__(self, filepath) -> None:
        self.filepath = filepath
        self.is_exists = path.exists(self.filepath)
    
    def create(self,headers_list):
        if path.exists(self.filepath):
            raise FileExistsError("もうある")
        try:
            with open(self.filepath, "wt", encoding="utf-8", newline="") as file_object:
                writer = csv.writer(file_object)
                writer.writerow(headers_list)
            return self.filepath
        except Exception as e:
            raise Exception(e)        


    def append(self, line):
        # 1行渡すとCSVに1行追記する 
        try:
            with open(self.filepath, "a", encoding="utf-8", newline="") as file_object:
                writer = csv.writer(file_object)
                writer.writerow(line)
            return self.filepath
        except Exception as e:
            raise Exception(e)

    def read(self):
        try:
            with open(self.filepath, "r") as file_object:
                reader = csv.reader(file_object)
                list_of_rows = list(reader)
                return list_of_rows
        except Exception as e:
            raise Exception(e)