Advanced Python Concepts: File Handling and Exception Handling

1. File Handling

File Handling dalam Python memungkinkan kita untuk membaca dari dan menulis ke file di sistem. Ini sangat penting untuk berbagai operasi seperti menyimpan data, membaca konfigurasi, dan memproses log.

Konsep Dasar:

• open(): Membuka file
• read(), readline(), readlines(): Membaca dari file
• write(), writelines(): Menulis ke file
• close(): Menutup file
• with statement: Mengelola konteks file (membuka dan menutup secara otomatis)

Common modes:

• 'r' - read
• 'w' - write (truncates the file if it exists)
• 'a' - append
• 'r+' - read and write

Contoh dasar:

a) Opening a file:

python
file = open('example.txt', 'r')  # 'r' for read mode

b) Reading from a file:

python
content = file.read()  # Read entire file
line = file.readline()  # Read a single line
lines = file.readlines()  # Read all lines into a list

c) Writing to a file:

python
file = open('example.txt', 'w')  # 'w' for write mode
file.write('Hello, World!')

d) Appending to a file:

python
file = open('example.txt', 'a')  # 'a' for append mode
file.write('New content')

e) Managing file metadata:

python
import os
file_size = os.path.getsize('example.txt')
last_modified = os.path.getmtime('example.txt')

python
# Membaca file
with open('data.txt', 'r') as file:
    content = file.read()

# Menulis ke file
with open('output.txt', 'w') as file:
    file.write('Hello, World!')

Practical Implementations in Various Industries:

1. Banking and Fintech:

• Transaction Logging: Record all financial transactions in a log file for auditing purposes.
• Customer Data Management: Store and retrieve customer information from files.

Example:

python
def log_transaction(transaction):
    with open('transactions.log', 'a') as file:
        file.write(f"{transaction.timestamp},{transaction.account_id},{transaction.amount}\n")

def get_customer_data(customer_id):
    with open('customers.csv', 'r') as file:
        for line in file:
            if line.startswith(customer_id):
                return line.strip().split(',')
    return None

2. E-commerce:

• Product Catalog Management: Store product information in CSV or JSON files.
• Order Processing: Write order details to files for record-keeping.

Example:

python
import json

def update_product_catalog(product):
    with open('products.json', 'r+') as file:
        catalog = json.load(file)
        catalog[product['id']] = product
        file.seek(0)
        json.dump(catalog, file, indent=2)
        file.truncate()

def process_order(order):
    with open('orders.log', 'a') as file:
        file.write(json.dumps(order) + '\n')

3. Healthcare:

• Patient Records: Store and retrieve patient medical histories from files.
• Appointment Scheduling: Manage appointment schedules using file-based storage.

Example:

python
def update_patient_record(patient_id, new_data):
    filename = f"patient_{patient_id}.json"
    with open(filename, 'r+') as file:
        record = json.load(file)
        record.update(new_data)
        file.seek(0)
        json.dump(record, file, indent=2)
        file.truncate()

def get_appointments(date):
    with open('appointments.csv', 'r') as file:
        return [line.strip().split(',') for line in file if line.startswith(date)]

4. Production and Inventory Management:

• Inventory Tracking: Maintain inventory levels in files.
• Production Logs: Record production data in log files.

Example:

python
def update_inventory(product_id, quantity_change):
    with open('inventory.csv', 'r+') as file:
        lines = file.readlines()
        for i, line in enumerate(lines):
            if line.startswith(product_id):
                current_quantity = int(line.strip().split(',')[1])
                new_quantity = current_quantity + quantity_change
                lines[i] = f"{product_id},{new_quantity}\n"
                break
        file.seek(0)
        file.writelines(lines)
        file.truncate()

def log_production(product_id, quantity):
    with open('production.log', 'a') as file:
        file.write(f"{datetime.now()},{product_id},{quantity}\n")

5. Logistics:

• Shipment Tracking: Store and update shipment statuses in files.
• Route Planning: Read route information from configuration files.

Example:

python
def update_shipment_status(shipment_id, status):
    with open('shipments.csv', 'r+') as file:
        lines = file.readlines()
        for i, line in enumerate(lines):
            if line.startswith(shipment_id):
                lines[i] = f"{shipment_id},{status},{datetime.now()}\n"
                break
        file.seek(0)
        file.writelines(lines)
        file.truncate()

def get_route_plan(route_id):
    with open('routes.json', 'r') as file:
        routes = json.load(file)
        return routes.get(route_id)

6. Maintenance:

• Equipment Logs: Record maintenance activities in log files.
• Scheduled Maintenance: Read maintenance schedules from files.

Example:

python
def log_maintenance(equipment_id, activity):
    with open('maintenance.log', 'a') as file:
        file.write(f"{datetime.now()},{equipment_id},{activity}\n")

def get_maintenance_schedule():
    with open('maintenance_schedule.csv', 'r') as file:
        return [line.strip().split(',') for line in file]

7. Project Management:

• Task Tracking: Store project tasks and their statuses in files.
• Project Reports: Generate reports by reading project data from files.

Example:

python
def update_task_status(project_id, task_id, status):
    filename = f"project_{project_id}.json"
    with open(filename, 'r+') as file:
        project_data = json.load(file)
        for task in project_data['tasks']:
            if task['id'] == task_id:
                task['status'] = status
                break
        file.seek(0)
        json.dump(project_data, file, indent=2)
        file.truncate()

def generate_project_report(project_id):
    filename = f"project_{project_id}.json"
    with open(filename, 'r') as file:
        project_data = json.load(file)
        # Process project_data and generate report
        return report

8. Quality Management:

• Quality Control Logs: Record quality control checks in log files.
• Compliance Reports: Generate compliance reports by reading quality data from files.

Example:

python
def log_quality_check(product_id, result):
    with open('quality_checks.log', 'a') as file:
        file.write(f"{datetime.now()},{product_id},{result}\n")

def generate_compliance_report(start_date, end_date):
    report_data = []
    with open('quality_checks.log', 'r') as file:
        for line in file:
            date, product_id, result = line.strip().split(',')
            if start_date <= date <= end_date:
                report_data.append((date, product_id, result))
    # Process report_data and generate report
    return report

2. Exception Handling

Exception Handling memungkinkan kita untuk menangani error yang mungkin terjadi saat runtime, memastikan program tetap berjalan atau berhenti dengan cara yang terkontrol.

Konsep Dasar:

• try: Blok kode yang mungkin menyebabkan exception
• except: Menangani exception spesifik
• else: Dieksekusi jika tidak ada exception
• finally: Selalu dieksekusi, baik ada exception atau tidak
• raise: Memunculkan exception secara manual

Syntax and Structure:

• Basic try-except block:

  python
  
  try:
      # Code that may raise an exception
  except ExceptionType:
      # Code to handle the exception
  
  

• Finally block:

  python
  
  try:
      # Code that may raise an exception
  except ExceptionType:
      # Code to handle the exception
  finally:
      # Code that will run no matter what
  

• Raising exceptions:

  python
  
  raise ExceptionType('Error message')
  

Advanced Techniques:

• Handling multiple exceptions:

  python
  
  try:
      # Code that may raise an exception
  except (ExceptionType1, ExceptionType2) as e:
      # Code to handle the exceptions
  

Contoh dasar:

python
try:
    result = 10 / 0
except ZeroDivisionError:
    print("Tidak bisa membagi dengan nol")
else:
    print("Hasil:", result)
finally:
    print("Operasi selesai")

Implementing Exception Handling:

1. Banking and Fintech:

python
def perform_transaction(account, amount):
    try:
        if amount > account.balance:
            raise ValueError("Insufficient funds")
        account.balance -= amount
        log_transaction(account, amount)
    except ValueError as e:
        print(f"Transaction failed: {e}")
        log_error(account, str(e))
    except IOError:
        print("Error logging transaction")
        # Implement rollback mechanism
    else:
        print("Transaction successful")
    finally:
        update_account_status(account)

2. E-commerce:

python
def process_order(order):
    try:
        validate_order(order)
        charge_payment(order)
        update_inventory(order)
        send_confirmation_email(order)
    except ValidationError as e:
        log_error(f"Order validation failed: {e}")
        notify_customer(order, "Invalid order")
    except PaymentError as e:
        log_error(f"Payment failed: {e}")
        notify_customer(order, "Payment issue")
    except InventoryError as e:
        log_error(f"Inventory update failed: {e}")
        refund_payment(order)
        notify_customer(order, "Out of stock")
    except EmailError:
        log_error("Failed to send confirmation email")
        # Order processed but email failed, handle accordingly
    else:
        log_success(f"Order {order.id} processed successfully")

3. Healthcare:

python
def update_patient_record(patient_id, new_data):
    try:
        record = fetch_patient_record(patient_id)
        validated_data = validate_medical_data(new_data)
        merged_record = merge_records(record, validated_data)
        save_patient_record(patient_id, merged_record)
    except PatientNotFoundError:
        log_error(f"Patient {patient_id} not found")
        raise
    except ValidationError as e:
        log_error(f"Invalid medical data: {e}")
        raise
    except DatabaseError as e:
        log_error(f"Database error: {e}")
        raise SystemError("Critical database error")
    else:
        log_audit(f"Updated record for patient {patient_id}")

4. Production and Inventory Management:

python
def update_inventory(product_id, quantity_change):
    try:
        current_quantity = get_current_inventory(product_id)
        new_quantity = current_quantity + quantity_change
        if new_quantity < 0:
            raise ValueError("Inventory cannot be negative")
        save_inventory_level(product_id, new_quantity)
    except ProductNotFoundError:
        log_error(f"Product {product_id} not found in inventory")
        raise
    except ValueError as e:
        log_error(f"Invalid inventory update: {e}")
        raise
    except DatabaseError as e:
        log_error(f"Failed to update inventory: {e}")
        raise SystemError("Critical inventory system error")
    else:
        log_inventory_change(product_id, quantity_change)

5. Logistics:

python
def update_shipment_status(shipment_id, new_status):
    try:
        shipment = get_shipment(shipment_id)
        validate_status_transition(shipment.status, new_status)
        update_status_in_database(shipment_id, new_status)
        notify_stakeholders(shipment_id, new_status)
    except ShipmentNotFoundError:
        log_error(f"Shipment {shipment_id} not found")
        raise
    except InvalidStatusTransitionError as e:
        log_error(f"Invalid status transition: {e}")
        raise
    except DatabaseError as e:
        log_error(f"Database error: {e}")
        raise SystemError("Critical shipment tracking system error")
    except NotificationError:
        log_warning(f"Failed to notify stakeholders for shipment {shipment_id}")
        # Continue processing as notification is non-critical
    else:
        log_shipment_update(shipment_id, new_status)

Implementasi FILE HANDLING dan EXCEPTION HANDLING dalam Berbagai Industri:

1. Perbankan dan Fintech:

python
import json

class BankAccount:
    def __init__(self, account_number, balance):
        self.account_number = account_number
        self.balance = balance

    def withdraw(self, amount):
        try:
            if amount > self.balance:
                raise ValueError("Dana tidak cukup")
            self.balance -= amount
            self.log_transaction("Penarikan", amount)
        except ValueError as e:
            print(f"Error: {e}")
            self.log_transaction("Penarikan Gagal", amount)

    def log_transaction(self, transaction_type, amount):
        try:
            with open(f"{self.account_number}_transactions.log", 'a') as file:
                log_entry = f"{transaction_type}: ${amount}\n"
                file.write(log_entry)
        except IOError:
            print("Gagal menulis log transaksi")

# Penggunaan
account = BankAccount("1234567890", 1000)
account.withdraw(500)
account.withdraw(2000)  # Ini akan menyebabkan error

2. E-commerce:

python
import csv
from datetime import datetime

class InventoryManager:
    def __init__(self, inventory_file):
        self.inventory_file = inventory_file
        self.inventory = self.load_inventory()

    def load_inventory(self):
        try:
            with open(self.inventory_file, 'r') as file:
                reader = csv.DictReader(file)
                return list(reader)
        except FileNotFoundError:
            print("File inventori tidak ditemukan. Membuat file baru.")
            return []

    def update_stock(self, product_id, quantity_change):
        for item in self.inventory:
            if item['product_id'] == product_id:
                new_quantity = int(item['quantity']) + quantity_change
                if new_quantity < 0:
                    raise ValueError("Stok tidak boleh negatif")
                item['quantity'] = str(new_quantity)
                self.save_inventory()
                return
        raise ValueError("Produk tidak ditemukan")

    def save_inventory(self):
        try:
            with open(self.inventory_file, 'w', newline='') as file:
                writer = csv.DictWriter(file, fieldnames=['product_id', 'name', 'quantity'])
                writer.writeheader()
                writer.writerows(self.inventory)
        except IOError:
            print("Gagal menyimpan inventori")

# Penggunaan
inventory_manager = InventoryManager('inventory.csv')
try:
    inventory_manager.update_stock('P001', -5)
except ValueError as e:
    print(f"Error: {e}")

3. Healthcare:

python
import json
from datetime import datetime

class PatientRecord:
    def __init__(self, patient_id):
        self.patient_id = patient_id
        self.record_file = f"patient_{patient_id}.json"

    def add_medical_entry(self, entry_type, details):
        try:
            with open(self.record_file, 'r+') as file:
                try:
                    records = json.load(file)
                except json.JSONDecodeError:
                    records = []
                
                new_entry = {
                    "type": entry_type,
                    "details": details,
                    "timestamp": datetime.now().isoformat()
                }
                records.append(new_entry)
                
                file.seek(0)
                json.dump(records, file, indent=2)
                file.truncate()
        except FileNotFoundError:
            with open(self.record_file, 'w') as file:
                json.dump([{
                    "type": entry_type,
                    "details": details,
                    "timestamp": datetime.now().isoformat()
                }], file, indent=2)

    def get_medical_history(self):
        try:
            with open(self.record_file, 'r') as file:
                return json.load(file)
        except FileNotFoundError:
            return []
        except json.JSONDecodeError:
            print("Error: File rusak")
            return []

# Penggunaan
patient = PatientRecord("12345")
patient.add_medical_entry("Diagnosis", "Flu")
print(patient.get_medical_history())

4. Production dan Inventory:

python
import csv
from datetime import datetime

class ProductionLog:
    def __init__(self, log_file):
        self.log_file = log_file

    def log_production(self, product_id, quantity):
        try:
            with open(self.log_file, 'a', newline='') as file:
                writer = csv.writer(file)
                writer.writerow([datetime.now(), product_id, quantity])
        except IOError:
            print("Gagal menulis log produksi")

    def get_daily_production(self, date):
        daily_production = {}
        try:
            with open(self.log_file, 'r') as file:
                reader = csv.reader(file)
                for row in reader:
                    log_date = datetime.strptime(row[0], '%Y-%m-%d %H:%M:%S.%f')
                    if log_date.date() == date:
                        product_id = row[1]
                        quantity = int(row[2])
                        daily_production[product_id] = daily_production.get(product_id, 0) + quantity
        except FileNotFoundError:
            print("File log tidak ditemukan")
        return daily_production

# Penggunaan
production_log = ProductionLog('production.csv')
production_log.log_production('P001', 100)
print(production_log.get_daily_production(datetime.now().date()))

5. Logistik:

python
import json
from datetime import datetime

class ShipmentTracker:
    def __init__(self, tracking_file):
        self.tracking_file = tracking_file

    def update_shipment_status(self, shipment_id, status):
        try:
            with open(self.tracking_file, 'r+') as file:
                try:
                    shipments = json.load(file)
                except json.JSONDecodeError:
                    shipments = {}

                shipments[shipment_id] = {
                    "status": status,
                    "last_updated": datetime.now().isoformat()
                }

                file.seek(0)
                json.dump(shipments, file, indent=2)
                file.truncate()
        except IOError:
            print("Gagal memperbarui status pengiriman")

    def get_shipment_status(self, shipment_id):
        try:
            with open(self.tracking_file, 'r') as file:
                shipments = json.load(file)
                return shipments.get(shipment_id, {"status": "Tidak ditemukan"})
        except FileNotFoundError:
            return {"status": "File tracking tidak ditemukan"}
        except json.JSONDecodeError:
            return {"status": "Error: File rusak"}

# Penggunaan
tracker = ShipmentTracker('shipments.json')
tracker.update_shipment_status('S001', 'In Transit')
print(tracker.get_shipment_status('S001'))

6. Maintenance:

python
import csv
from datetime import datetime, timedelta

class MaintenanceScheduler:
    def __init__(self, schedule_file):
        self.schedule_file = schedule_file

    def schedule_maintenance(self, equipment_id, maintenance_date):
        try:
            with open(self.schedule_file, 'a', newline='') as file:
                writer = csv.writer(file)
                writer.writerow([equipment_id, maintenance_date.isoformat()])
        except IOError:
            print("Gagal menjadwalkan pemeliharaan")

    def get_upcoming_maintenance(self, days=7):
        upcoming = []
        cutoff_date = datetime.now() + timedelta(days=days)
        try:
            with open(self.schedule_file, 'r') as file:
                reader = csv.reader(file)
                for row in reader:
                    equipment_id, date_str = row
                    maintenance_date = datetime.fromisoformat(date_str)
                    if datetime.now() <= maintenance_date <= cutoff_date:
                        upcoming.append((equipment_id, maintenance_date))
        except FileNotFoundError:
            print("File jadwal tidak ditemukan")
        return upcoming

# Penggunaan
scheduler = MaintenanceScheduler('maintenance_schedule.csv')
scheduler.schedule_maintenance('EQ001', datetime.now() + timedelta(days=5))
print(scheduler.get_upcoming_maintenance())

7. Project Management:

python
import json
from datetime import datetime

class ProjectManager:
    def __init__(self, project_file):
        self.project_file = project_file

    def add_task(self, project_id, task_name, deadline):
        try:
            with open(self.project_file, 'r+') as file:
                try:
                    projects = json.load(file)
                except json.JSONDecodeError:
                    projects = {}

                if project_id not in projects:
                    projects[project_id] = {"tasks": []}

                projects[project_id]["tasks"].append({
                    "name": task_name,
                    "deadline": deadline.isoformat(),
                    "status": "Pending"
                })

                file.seek(0)
                json.dump(projects, file, indent=2)
                file.truncate()
        except IOError:
            print("Gagal menambahkan tugas")

    def get_project_status(self, project_id):
        try:
            with open(self.project_file, 'r') as file:
                projects = json.load(file)
                return projects.get(project_id, {"status": "Proyek tidak ditemukan"})
        except FileNotFoundError:
            return {"status": "File proyek tidak ditemukan"}
        except json.JSONDecodeError:
            return {"status": "Error: File rusak"}

# Penggunaan
pm = ProjectManager('projects.json')
pm.add_task('P001', 'Desain UI', datetime.now() + timedelta(days=10))
print(pm.get_project_status('P001'))

8. Quality Management:

python
import csv
from datetime import datetime

class QualityControl:
    def __init__(self, qc_log_file):
        self.qc_log_file = qc_log_file

    def log_inspection(self, product_id, pass_status, inspector, notes):
        try:
            with open(self.qc_log_file, 'a', newline='') as file:
                writer = csv.writer(file)
                writer.writerow([datetime.now(), product_id, pass_status, inspector, notes])
        except IOError:
            print("Gagal mencatat inspeksi")

    def get_failure_rate(self, start_date, end_date):
        total = 0
        failures = 0
        try:
            with open(self.qc_log_file, 'r') as file:
                reader = csv.reader(file)
                for row in reader:
                    inspection_date = datetime.strptime(row[0], '%Y-%m-%d %H:%M:%S.%f')
                    if start_date <= inspection_date <= end_date:
                        total += 1
                        if row[2].lower() == 'fail':
                            failures += 1
        except FileNotFoundError:
            print("File log QC tidak ditemukan")
        
        return failures / total if total > 0 else 0

# Penggunaan
qc = QualityControl('qc_log.csv')
qc.log_inspection('P001', 'Pass', 'John Doe', 'No issues')
print(qc.get_failure_rate(datetime.now() - timedelta(days=30), datetime.now()))

Contoh-contoh di atas menunjukkan bagaimana File Handling dan Exception Handling dapat diimplementasikan dalam berbagai aplikasi industri. Beberapa poin penting:

1. File Handling digunakan untuk menyimpan dan membaca data penting seperti log transaksi, inventori, rekam medis, jadwal produksi, dan laporan kualitas.
2. Exception Handling memastikan bahwa program dapat menangani situasi seperti file yang tidak ditemukan, format data yang tidak valid, atau operasi yang tidak diizinkan (misalnya, penarikan melebihi saldo).
3. Penggunaan context manager (with statement) untuk mengelola file secara aman dan efisien.
4. Pemisahan logika bisnis dari operasi file, memungkinkan fleksibilitas dan maintainability yang lebih baik.
5. Penanganan berbagai jenis file (CSV, JSON) sesuai dengan kebutuhan aplikasi.

Dengan mengimplementasikan File Handling dan Exception Handling dengan baik, aplikasi menjadi lebih robust, dapat diandalkan, dan mampu menangani berbagai skenario yang mungkin terjadi dalam operasi sehari-hari di berbagai industri.