Master Python file operations, reading and writing data, working with different file formats, and handling file system operations safely and efficiently.
File handling is a fundamental skill in Python programming that allows you to work with external data, create logs, store user information, and process files. Python provides built-in functions and methods to handle various file operations efficiently and safely. In this lesson, we'll explore how to read from and write to files, work with different file formats, and implement proper error handling to make your file operations robust and reliable.
The foundation of file handling in Python starts with opening files using the built-in open() function. Python supports various file modes that determine how you can interact with the file:
The most important concept is using the context manager (with statement) to ensure files are properly closed:
1# Proper way using context manager
2with open('filename.txt', 'r') as file:
3 content = file.read()
4 # File automatically closed when exiting the with block
5
6# Manual way (not recommended)
7file = open('filename.txt', 'r')
8content = file.read()
9file.close() # Must remember to closeThe context manager ensures that files are always properly closed, even if an error occurs during processing.
Python provides several methods for reading file content, each suited for different scenarios:
read(): Reads the entire file content as a single string
1with open('data.txt', 'r') as file:
2 content = file.read() # Entire file as stringreadline(): Reads one line at a time
1with open('data.txt', 'r') as file:
2 first_line = file.readline()
3 second_line = file.readline()readlines(): Reads all lines into a list
1with open('data.txt', 'r') as file:
2 lines = file.readlines() # List of stringsIterating over file objects: Memory-efficient for large files
1with open('large_file.txt', 'r') as file:
2 for line in file:
3 process_line(line) # Process one line at a timeFor structured data like CSV files, you can use the csv module or pandas for more advanced processing.
Writing files in Python is straightforward with several methods available depending on your needs:
write(): Writes a string to the file
1with open('output.txt', 'w') as file:
2 file.write('Hello, World!\n')
3 file.write('Second line')writelines(): Writes a list of strings to the file
1lines = ['Line 1\n', 'Line 2\n', 'Line 3\n']
2with open('output.txt', 'w') as file:
3 file.writelines(lines)Appending to files: Adding content without overwriting
1with open('log.txt', 'a') as file:
2 file.write('New log entry\n')Working with CSV data: Structured data handling
1import csv
2
3# Writing CSV
4with open('data.csv', 'w', newline='') as file:
5 writer = csv.writer(file)
6 writer.writerow(['Name', 'Age', 'City'])
7 writer.writerow(['Alice', 30, 'New York'])
8
9# Reading CSV
10with open('data.csv', 'r') as file:
11 reader = csv.reader(file)
12 for row in reader:
13 print(row)File operations can fail for various reasons, so proper error handling is crucial for robust applications:
Common file exceptions:
Exception handling pattern:
1import os
2
3def safe_file_read(filename):
4 try:
5 with open(filename, 'r') as file:
6 return file.read()
7 except FileNotFoundError:
8 print(f"File {filename} not found")
9 return None
10 except PermissionError:
11 print(f"Permission denied: {filename}")
12 return None
13 except IOError as e:
14 print(f"Error reading file: {e}")
15 return NoneBest practices:
File existence and path operations:
1import os.path
2
3if os.path.exists('myfile.txt'):
4 with open('myfile.txt', 'r') as file:
5 content = file.read()
6else:
7 print("File does not exist")Build a comprehensive log file analyzer that reads server logs, extracts important information, generates reports, and handles errors gracefully.
1# Log File Analyzer System
2import csv
3import os
4import re
5from datetime import datetime
6from collections import defaultdict
7
8class LogAnalyzer:
9 def __init__(self, log_directory='logs'):
10 self.log_directory = log_directory
11 self.analysis_results = {}
12 self.ensure_directory_exists()
13
14 def ensure_directory_exists(self):
15 """Create log directory if it doesn't exist"""
16 if not os.path.exists(self.log_directory):
17 os.makedirs(self.log_directory)
18 print(f"Created directory: {self.log_directory}")
19
20 def create_sample_log(self, filename='server.log'):
21 """Create a sample log file for demonstration"""
22 sample_logs = [
23 "2024-01-15 10:30:45 INFO User login successful: user123",
24 "2024-01-15 10:31:12 ERROR Database connection failed",
25 "2024-01-15 10:31:45 WARNING High memory usage detected",
26 "2024-01-15 10:32:15 INFO User logout: user123",
27 "2024-01-15 10:33:20 ERROR File not found: config.xml",
28 "2024-01-15 10:34:10 INFO New user registered: user456",
29 "2024-01-15 10:35:05 WARNING Disk space low",
30 "2024-01-15 10:36:30 ERROR Authentication failed for user789"
31 ]
32
33 filepath = os.path.join(self.log_directory, filename)
34 try:
35 with open(filepath, 'w') as file:
36 for log_entry in sample_logs:
37 file.write(log_entry + '\n')
38 print(f"Sample log file created: {filepath}")
39 except IOError as e:
40 print(f"Error creating sample log: {e}")
41
42 def parse_log_file(self, filename):
43 """Parse log file and extract structured information"""
44 filepath = os.path.join(self.log_directory, filename)
45
46 if not os.path.exists(filepath):
47 print(f"Log file not found: {filepath}")
48 return None
49
50 log_entries = []
51 log_pattern = r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) (INFO|ERROR|WARNING) (.+)'
52
53 try:
54 with open(filepath, 'r') as file:
55 for line_number, line in enumerate(file, 1):
56 line = line.strip()
57 if line:
58 match = re.match(log_pattern, line)
59 if match:
60 timestamp, level, message = match.groups()
61 log_entries.append({
62 'line_number': line_number,
63 'timestamp': timestamp,
64 'level': level,
65 'message': message
66 })
67 else:
68 print(f"Invalid log format at line {line_number}: {line}")
69
70 print(f"Successfully parsed {len(log_entries)} log entries from {filename}")
71 return log_entries
72
73 except FileNotFoundError:
74 print(f"File not found: {filepath}")
75 return None
76 except PermissionError:
77 print(f"Permission denied reading: {filepath}")
78 return None
79 except IOError as e:
80 print(f"Error reading log file: {e}")
81 return None
82
83 def analyze_logs(self, log_entries):
84 """Analyze log entries and generate statistics"""
85 if not log_entries:
86 return None
87
88 analysis = {
89 'total_entries': len(log_entries),
90 'level_counts': defaultdict(int),
91 'error_messages': [],
92 'warning_messages': [],
93 'time_range': {
94 'start': log_entries[0]['timestamp'],
95 'end': log_entries[-1]['timestamp']
96 },
97 'user_activities': []
98 }
99
100 for entry in log_entries:
101 level = entry['level']
102 message = entry['message']
103
104 analysis['level_counts'][level] += 1
105
106 if level == 'ERROR':
107 analysis['error_messages'].append({
108 'timestamp': entry['timestamp'],
109 'message': message,
110 'line': entry['line_number']
111 })
112 elif level == 'WARNING':
113 analysis['warning_messages'].append({
114 'timestamp': entry['timestamp'],
115 'message': message,
116 'line': entry['line_number']
117 })
118
119 # Extract user activities
120 if 'user' in message.lower():
121 analysis['user_activities'].append({
122 'timestamp': entry['timestamp'],
123 'activity': message
124 })
125
126 return analysis
127
128 def generate_report(self, analysis, output_filename='log_report.txt'):
129 """Generate a comprehensive report file"""
130 if not analysis:
131 print("No analysis data to generate report")
132 return
133
134 report_path = os.path.join(self.log_directory, output_filename)
135
136 try:
137 with open(report_path, 'w') as report_file:
138 report_file.write("LOG ANALYSIS REPORT\n")
139 report_file.write("=" * 50 + "\n\n")
140
141 # Summary section
142 report_file.write(f"Analysis Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
143 report_file.write(f"Total Log Entries: {analysis['total_entries']}\n")
144 report_file.write(f"Time Range: {analysis['time_range']['start']} to {analysis['time_range']['end']}\n\n")
145
146 # Level distribution
147 report_file.write("LOG LEVEL DISTRIBUTION\n")
148 report_file.write("-" * 25 + "\n")
149 for level, count in analysis['level_counts'].items():
150 percentage = (count / analysis['total_entries']) * 100
151 report_file.write(f"{level}: {count} ({percentage:.1f}%)\n")
152 report_file.write("\n")
153
154 # Error summary
155 if analysis['error_messages']:
156 report_file.write(f"ERRORS FOUND ({len(analysis['error_messages'])})\n")
157 report_file.write("-" * 20 + "\n")
158 for error in analysis['error_messages']:
159 report_file.write(f"Line {error['line']}: [{error['timestamp']}] {error['message']}\n")
160 report_file.write("\n")
161
162 # Warning summary
163 if analysis['warning_messages']:
164 report_file.write(f"WARNINGS FOUND ({len(analysis['warning_messages'])})\n")
165 report_file.write("-" * 22 + "\n")
166 for warning in analysis['warning_messages']:
167 report_file.write(f"Line {warning['line']}: [{warning['timestamp']}] {warning['message']}\n")
168 report_file.write("\n")
169
170 # User activity summary
171 if analysis['user_activities']:
172 report_file.write(f"USER ACTIVITIES ({len(analysis['user_activities'])})\n")
173 report_file.write("-" * 20 + "\n")
174 for activity in analysis['user_activities']:
175 report_file.write(f"[{activity['timestamp']}] {activity['activity']}\n")
176
177 print(f"Report generated successfully: {report_path}")
178
179 except IOError as e:
180 print(f"Error generating report: {e}")
181
182 def export_to_csv(self, log_entries, csv_filename='log_data.csv'):
183 """Export parsed log data to CSV format"""
184 if not log_entries:
185 print("No log entries to export")
186 return
187
188 csv_path = os.path.join(self.log_directory, csv_filename)
189
190 try:
191 with open(csv_path, 'w', newline='') as csvfile:
192 fieldnames = ['line_number', 'timestamp', 'level', 'message']
193 writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
194
195 writer.writeheader()
196 for entry in log_entries:
197 writer.writerow(entry)
198
199 print(f"Log data exported to CSV: {csv_path}")
200
201 except IOError as e:
202 print(f"Error exporting to CSV: {e}")
203
204 def run_full_analysis(self, log_filename='server.log'):
205 """Run complete analysis workflow"""
206 print("Starting Log File Analysis...")
207 print("=" * 40)
208
209 # Create sample log if it doesn't exist
210 log_path = os.path.join(self.log_directory, log_filename)
211 if not os.path.exists(log_path):
212 print("Sample log file not found. Creating one...")
213 self.create_sample_log(log_filename)
214
215 # Parse log file
216 print(f"\nParsing log file: {log_filename}")
217 log_entries = self.parse_log_file(log_filename)
218
219 if not log_entries:
220 print("Failed to parse log file. Exiting.")
221 return
222
223 # Analyze parsed data
224 print("\nAnalyzing log data...")
225 analysis = self.analyze_logs(log_entries)
226
227 # Generate reports
228 print("\nGenerating reports...")
229 self.generate_report(analysis)
230 self.export_to_csv(log_entries)
231
232 # Display summary
233 print("\nANALYSIS SUMMARY")
234 print("-" * 20)
235 print(f"Total entries processed: {analysis['total_entries']}")
236 print(f"Errors found: {len(analysis['error_messages'])}")
237 print(f"Warnings found: {len(analysis['warning_messages'])}")
238 print(f"User activities: {len(analysis['user_activities'])}")
239
240 print("\nAnalysis complete! Check the logs directory for generated reports.")
241
242# Demonstrate the log analyzer
243if __name__ == "__main__":
244 analyzer = LogAnalyzer()
245 analyzer.run_full_analysis()
246
247 print("\n" + "="*50)
248 print("ADDITIONAL FILE OPERATIONS DEMO")
249 print("="*50)
250
251 # Demonstrate various file operations
252 demo_file = os.path.join('logs', 'demo.txt')
253
254 # Writing different types of content
255 try:
256 with open(demo_file, 'w') as f:
257 f.write("File Handling Demonstration\n")
258 f.write("=" * 30 + "\n")
259 f.write("Line 1: Simple text\n")
260 f.write("Line 2: Numbers: 123, 456\n")
261 f.write("Line 3: Special chars: @#$%^&*\n")
262
263 print(f"Demo file created: {demo_file}")
264
265 # Reading and displaying content
266 with open(demo_file, 'r') as f:
267 print("\nFile content:")
268 print("-" * 20)
269 print(f.read())
270
271 # Appending content
272 with open(demo_file, 'a') as f:
273 f.write("Line 4: Appended content\n")
274 f.write(f"Line 5: Timestamp: {datetime.now()}\n")
275
276 print("Content appended successfully")
277
278 # Reading line by line
279 print("\nReading line by line:")
280 print("-" * 20)
281 with open(demo_file, 'r') as f:
282 for i, line in enumerate(f, 1):
283 print(f"Line {i}: {line.strip()}")
284
285 except IOError as e:
286 print(f"Error in demo: {e}")
287
288 print("\nFile handling demonstration complete!")Your progress is saved in this browser, on this device only.