def adjust_ratios(amounts):
"""
Adjusts the ratios of given amounts so that their total becomes 100%.
Args:
amounts (list of float): A list of amounts to be adjusted.
Returns:
list of float: Adjusted ratios rounded to two decimal places.
Raises:
ValueError: If any of the amounts are negative.
"""
# Validate input amounts
if any(amount < 0 for amount in amounts):
raise ValueError("All amounts must be non-negative.")
# Calculate the ratios
total_amount = sum(amounts)
ratios = [amount / total_amount for amount in amounts]
# Round the ratios to two decimal places and calculate the rounded total
rounded_ratios = [round(ratio * 100, 2) for ratio in ratios]
rounded_total = sum(rounded_ratios)
# Adjust the difference to ensure the total is exactly 100%
difference = 100 - rounded_total
max_ratio_index = rounded_ratios.index(max(rounded_ratios))
rounded_ratios[max_ratio_index] += difference
return rounded_ratios# データベース格納先ディレクトリとファイル名
SQLITE_CONFIG = {
'db_dir': 'db',
'db_filename': 'demo.db'
}
import os
def create_directory_structure(project_name):
# プロジェクトのルートディレクトリを作成
os.makedirs(project_name)
# srcディレクトリを作成
src_dir = os.path.join(project_name, 'src')
os.makedirs(src_dir)
# app.py, database.py, utils.pyを作成
open(os.path.join(src_dir, 'app.py'), 'a').close()
open(os.path.join(src_dir, 'database.py'), 'a').close()
open(os.path.join(src_dir, 'utils.py'), 'a').close()
# testsディレクトリを作成
tests_dir = os.path.join(project_name, 'tests')
os.makedirs(tests_dir)
# test_database.py, test_utils.pyを作成
open(os.path.join(tests_dir, 'test_database.py'), 'a').close()
open(os.path.join(tests_dir, 'test_utils.py'), 'a').close()
# dbディレクトリを作成
db_dir = os.path.join(project_name, 'db')
os.makedirs(db_dir)
# test_database.dbを作成
open(os.path.join(db_dir, 'test_database.db'), 'a').close()
# requirements.txtを作成
with open(os.path.join(project_name, 'requirements.txt'), 'w') as f:
f.write('# プロジェクトの依存関係を記述\n')
# README.mdを作成
with open(os.path.join(project_name, 'README.md'), 'w') as f:
f.write('# ' + project_name + '\n')
f.write('プロジェクトの概要や使い方をここに記述\n')
# .gitignoreを作成
with open(os.path.join(project_name, '.gitignore'), 'w') as f:
f.write('# Gitの管理下に置かれるべきでないファイルやディレクトリを指定\n')
f.write('*.pyc\n')
f.write('__pycache__/\n')
f.write('db/*.db\n')
if __name__ == "__main__":
project_name = "backend_project"
create_directory_structure(project_name)
print(f"ディレクトリ構造 '{project_name}' が作成されました。")
import sqlite3
import pandas as pd
def generate_table_definitions_with_indexes(database_file, output_excel):
# SQLiteデータベースに接続
conn = sqlite3.connect(database_file)
cursor = conn.cursor()
# テーブルの一覧を取得
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
tables = cursor.fetchall()
# テーブル定義情報用のDataFrameを作成
table_definitions = pd.DataFrame(columns=['Table Name', 'Column Name', 'Data Type', 'Primary Key'])
# インデックス情報用のDataFrameを作成
index_definitions = pd.DataFrame(columns=['Table Name', 'Index Name', 'Column Name'])
for table in tables:
table_name = table[0]
# テーブルのカラム情報を取得
cursor.execute(f"PRAGMA table_info({table_name})")
table_info = cursor.fetchall()
for column_info in table_info:
column_name = column_info[1]
data_type = column_info[2]
is_primary_key = "Yes" if column_info[5] == 1 else "No"
table_definitions = pd.concat([table_definitions, pd.DataFrame({
'Table Name': [table_name],
'Column Name': [column_name],
'Data Type': [data_type],
'Primary Key': [is_primary_key]
})], ignore_index=True)
# テーブルのインデックス情報を取得
cursor.execute(f"PRAGMA index_list({table_name})")
index_list = cursor.fetchall()
for index_info in index_list:
index_name = index_info[1]
# インデックスに含まれるカラムを取得
cursor.execute(f"PRAGMA index_info({index_name})")
index_info_details = cursor.fetchall()
for detail in index_info_details:
column_name = detail[2] # インデックスに含まれるカラム名
index_definitions = pd.concat([index_definitions, pd.DataFrame({
'Table Name': [table_name],
'Index Name': [index_name],
'Column Name': [column_name]
})], ignore_index=True)
# Excelファイルに書き出し（テーブル定義とインデックス情報を別のシートに保存）
with pd.ExcelWriter(output_excel) as writer:
table_definitions.to_excel(writer, index=False, sheet_name='Table Definitions')
index_definitions.to_excel(writer, index=False, sheet_name='Index Definitions')
# 接続を閉じる
conn.close()
print(f"テーブル定義書とインデックス情報を {output_excel} に出力しました。")
# データベースファイルと出力先Excelファイルを指定して関数を呼び出し
generate_table_definitions_with_indexes('demo.db', 'table_definitions.xlsx')
# lib_log_manager.py
import logging
from logging.handlers import RotatingFileHandler
class LogManager:
@staticmethod
def setup_logging(log_file='application.log', level=logging.INFO):
"""
ログシステムの初期設定を行います。
:param log_file: ログファイルの名前
:param level: ログレベル
"""
log_format = '%(asctime)s - %(levelname)s - %(message)s'
logging.basicConfig(level=level,
format=log_format,
datefmt='%Y-%m-%d %H:%M:%S')
# ファイル出力を設定
if log_file:
try:
handler = RotatingFileHandler(log_file, maxBytes=1048576, backupCount=5)
handler.setFormatter(logging.Formatter(log_format))
logging.getLogger('').addHandler(handler)
except IOError as e:
print("Error setting up logging: ", e)
@staticmethod
def set_log_level(level):
"""
ログレベルを設定します。
:param level: ログレベル
"""
logging.getLogger('').setLevel(level)
@staticmethod
def log_info(message):
logging.info(message)
@staticmethod
def log_error(message, include_traceback=False):
if include_traceback:
logging.error(message, exc_info=True)
else:
logging.error(message)
@staticmethod
def log_warning(message):
logging.warning(message)# database_handler.py
import sqlite3
import pandas as pd
from lib_log_manager import LogManager
from utils import get_db_path
from config import SQLITE_CONFIG
class SqliteHandler:
def __init__(self):
self.db_file = get_db_path(SQLITE_CONFIG['db_dir'], SQLITE_CONFIG['db_filename'])
self.conn = None
self.connect_to_database()
def connect_to_database(self):
try:
self.conn = sqlite3.connect(self.db_file)
LogManager.log_info("Database connection successfully established.")
except sqlite3.Error as e:
LogManager.log_error(f"Failed to connect to the database. Error: {e}")
# ここでエラーを再発生させるか、あるいは呼び出し元が対応できるように、Falseを返すこともできます。
raise e # エラーを再発生させる場合
# return False # 呼び出し元がエラー4を処理できるようにする場合
def fetch_table_data(self, table_name):
query = f"SELECT count(*) FROM {table_name};"
cursor = self.conn.cursor()
cursor.execute(query)
columns = [desc[0] for desc in cursor.description]
data = cursor.fetchall()
cursor.close()
df = pd.DataFrame(data, columns=columns)
return df
def insert_data_to_table(self, table_name, data):
cursor = self.conn.cursor()
for _, row in data.iterrows():
placeholders = ','.join(['?'] * len(row))
query = f"INSERT INTO {table_name} VALUES ({placeholders})"
cursor.execute(query, tuple(row))
self.conn.commit()
cursor.close()
def close_connection(self):
if self.conn:
self.conn.close()
# main.py
import os
from lib_sqlite_handler import SqliteHandler
import logging
from lib_log_manager import LogManager
def main():
try:
# データベースハンドラを作成し、データベースに接続
db = SqliteHandler()
# ソーステーブルからデータを取得して表示
source_table_name = 't_employees'
table_data = db.fetch_table_data(source_table_name)
LogManager.log_info("Data from SQLite table:")
LogManager.log_info(table_data)
# データベースとの接続を閉じる
db.close_connection()
except Exception as e:
print("An error occurred:", e)
if __name__ == "__main__":
LogManager.setup_logging(log_file='example.log', level=logging.INFO)
main()
import pytest
# test_adjust_ratios_to_100.pyの内容
import sys
import os
# testsディレクトリからの相対パスでtoolディレクトリへのパスを追加
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../tool')))
from adjustRatiosTo100.adjust_ratios_to_100 import adjust_ratios
# 正常系のテストケース
@pytest.mark.parametrize("input_amounts, expected_output", [
([217, 32, 500], [28.97, 4.27, 66.76]), # 既存の例
([100, 100, 100], [33.33, 33.33, 33.34]), # 均等な場合
([0, 0, 1000], [0, 0, 100]), # ゼロが含まれる場合
([1, 1, 1], [33.33, 33.33, 33.34]), # 小さい値で均等な場合
([1000], [100]), # 単一の値の場合
])
def test_adjust_ratios(input_amounts, expected_output):
adjusted = adjust_ratios(input_amounts)
# 小数点第二位までの比較で許容誤差を考慮
assert all(abs(a - b) < 0.01 for a, b in zip(adjusted, expected_output))
# 合計が100%を超える場合のテストケース（異常系）
def test_sum_over_100():
with pytest.raises(ValueError):
adjust_ratios([101, 101, 101])
# 入力値が負の場合のテストケース（異常系）
def test_negative_values():
with pytest.raises(ValueError):
adjust_ratios([-1, 200, 300])
import pytest
import os
import sys
# testsディレクトリからの相対パスでtoolディレクトリへのパスを追加
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../')))
from lib_log_manager import LogManager
import logging
@pytest.fixture
def logger(caplog):
caplog.set_level(logging.INFO)
return caplog
def test_setup_logging(caplog):
LogManager.setup_logging(log_file='test.log')
assert 'Log system initialized successfully.' in caplog.text
def test_set_log_level(caplog):
LogManager.set_log_level(logging.DEBUG)
assert 'Log level set to DEBUG.' in caplog.text
def test_log_info(caplog):
LogManager.log_info('This is an info message.')
assert 'This is an info message.' in caplog.text
def test_log_error(caplog):
LogManager.log_error('This is an error message.', include_traceback=True)
assert 'This is an error message.' in caplog.text
def test_log_warning(caplog):
LogManager.log_warning('This is a warning message.')
assert 'This is a warning message.' in caplog.textimport os
def get_db_path(dir_name,db_name):
"""
現在のスクリプトのディレクトリからデータベースファイルへの絶対パスを返す。
:param db_name: データベースファイルの名前（例: 'demo.db'）
:return: データベースファイルへの絶対パス
"""
current_dir = os.path.dirname(os.path.abspath(__file__))
db_path = os.path.join(current_dir, dir_name, db_name)
return db_path
