turns-00050.parquet:2570
0ec84b3bb6d5f50c9cb1fd0ddegenerate_repetitionAbsentFinal dense release
USER
Implement graceful shutdown in this script
import asyncio
import aiohttp
from aiohttp import web
from bs4 import BeautifulSoup
import json
from typing import List, Dict, Optional
import ssl
import logging
from datetime import datetime
import colorlog
import socket
import netifaces # Newly added
import argparse # Newly added
import sys # Newly added
Configuration settings
CONFIG = {
'base_url': 'https://10.0.0.2',
'username': 'misael_campos',
'password': 'd337O8AetR3p4',
'post_endpoint': '/action/pononuopticalinfo.html',
'who': 100,
'onuid': 0,
'pon_range': range(1, 17),
'onu_group_range': range(0, 2),
'request_timeout': aiohttp.ClientTimeout(
total=240, # Total operation timeout
connect=120, # Connection timeout
sock_read=120 # Socket read timeout
),
'max_concurrent_requests': 8, # Reduced to prevent server overload
'max_retries': 3, # Maximum number of retry attempts
'retry_delay': 1, # Delay between retries in seconds
'batch_size': 8, # Number of concurrent requests per batch
'host': '0.0.0.0', # Host address
'port': 8080, # Port number
'endpoint': '/data', # Added endpoint path
'ignore_row_length_warning': False # New configuration option
}
Configure logging
handler = colorlog.StreamHandler()
handler.setFormatter(colorlog.ColoredFormatter(
'%(log_color)s%(asctime)s [%(levelname)s] %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
log_colors={
'DEBUG': 'light_cyan',
'INFO': 'light_green',
'WARNING': 'light_yellow',
'ERROR': 'light_red',
'CRITICAL': 'light_red,bg_white',
}
))
logger = colorlog.getLogger(name)
logger.addHandler(handler)
logger.setLevel(logging.INFO)
Remove any existing handlers to avoid duplicate logs
for hdlr in logger.handlers[:-1]:
logger.removeHandler(hdlr)
Function to retrieve all IPv4 addresses
def get_all_ip_addresses() -> List[str]:
ips = []
interfaces = netifaces.interfaces()
for iface in interfaces:
addrs = netifaces.ifaddresses(iface)
ipv4 = addrs.get(netifaces.AF_INET)
if ipv4:
for addr in ipv4:
ip = addr.get('addr')
if ip and ip != '127.0.0.1':
ips.append(ip)
return ips
class AsyncScraper:
def init(self, base_url: str):
self.base_url = base_url.rstrip('/')
self.ssl_context = ssl.create_default_context()
self.ssl_context.check_hostname = False
self.ssl_context.verify_mode = ssl.CERT_NONE
self.session = None
self.sem = asyncio.Semaphore(CONFIG['max_concurrent_requests'])
async def __aenter__(self):
timeout = CONFIG['request_timeout']
self.session = aiohttp.ClientSession(
connector=aiohttp.TCPConnector(
ssl=self.ssl_context,
limit=CONFIG['max_concurrent_requests']
),
timeout=timeout,
headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'}
)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
await self.session.close()
async def login(self, username: str, password: str) -> bool:
login_url = f"{self.base_url}/action/main.html"
login_data = {
'user': username,
'pass': password,
'button': 'Login',
'who': '100',
}
for attempt in range(CONFIG['max_retries']):
try:
async with self.session.post(login_url, data=login_data) as response:
text = await response.text()
if "Sorry, you do not have access" in text:
logger.error("Login failed: Access denied")
return False
elif "Error" in text or "Incorrect password" in text:
logger.error("Login failed: Incorrect credentials")
return False
logger.info("Login successful!")
return True
except Exception as e:
logger.error(f"Login attempt {attempt + 1} failed: {e}")
if attempt < CONFIG['max_retries'] - 1:
await asyncio.sleep(CONFIG['retry_delay'])
else:
return False
async def fetch_pon_onu_optical_info(self, pon: int, onu_group: int) -> Optional[str]:
async with self.sem:
post_url = f"{self.base_url}{CONFIG['post_endpoint']}"
payload = {
'pon': pon,
'onu_group': onu_group,
'who': CONFIG['who'],
'onuid': CONFIG['onuid']
}
for attempt in range(CONFIG['max_retries']):
try:
async with self.session.post(post_url, data=payload) as response:
if response.status == 200:
content = await response.text()
logger.info(f"Successfully fetched data for PON {pon}, ONU Group {onu_group} (Attempt {attempt + 1})")
return content
else:
logger.warning(f"Error {response.status} for PON {pon}, ONU Group {onu_group} (Attempt {attempt + 1})")
except asyncio.TimeoutError:
logger.warning(f"Timeout fetching PON {pon}, ONU Group {onu_group} (Attempt {attempt + 1})")
except Exception as e:
logger.warning(f"Error fetching PON {pon}, ONU Group {onu_group} (Attempt {attempt + 1}): {e}")
if attempt < CONFIG['max_retries'] - 1:
await asyncio.sleep(CONFIG['retry_delay'])
logger.info(f"Retrying PON {pon}, ONU Group {onu_group}...")
else:
logger.error(f"All attempts failed for PON {pon}, ONU Group {onu_group}")
return None
def parse_html_table(html_content: str) -> List[Dict[str, Optional[str]]]:
if not html_content:
return []
soup = BeautifulSoup(html_content, 'html.parser')
table = soup.find('table', border="1", cellpadding="4", cellspacing="0")
if not table:
logger.warning("No table found in the HTML content")
return []
headers = []
header_row = table.find('tr')
if not header_row:
logger.warning("No header row found in the table")
return []
for th in header_row.find_all(['td', 'th']):
header_text = th.get_text(strip=True)
headers.append(header_text if header_text else f"Column_{len(headers)+1}")
data = []
for row in table.find_all('tr')[1:]:
cells = row.find_all('td')
if len(cells) != len(headers):
if not CONFIG['ignore_row_length_warning']:
logger.warning("Row does not match header length. Skipping row")
continue
row_data = {
header: cell.get_text(strip=True) or None
for header, cell in zip(headers, cells)
}
data.append(row_data)
return data
async def fetch_and_parse(scraper: AsyncScraper, pon: int, onu_group: int) -> Dict[int, Dict[int, List[Dict[str, Optional[str]]]]]:
html_content = await scraper.fetch_pon_onu_optical_info(pon, onu_group)
parsed_data = parse_html_table(html_content) if html_content else []
if not parsed_data:
logger.warning(f"No data parsed for PON {pon}, ONU Group {onu_group}")
return {pon: {onu_group: parsed_data}}
async def process_batch(scraper: AsyncScraper, batch: List[tuple]) -> List[Dict]:
tasks = [fetch_and_parse(scraper, pon, onu_group) for pon, onu_group in batch]
return await asyncio.gather(*tasks, return_exceptions=True)
async def collect_data(scraper: AsyncScraper) -> Dict[int, Dict[int, List[Dict[str, Optional[str]]]]]:
# Create all combinations of PON and ONU groups
all_combinations = [
(pon, onu_group)
for pon in CONFIG['pon_range']
for onu_group in CONFIG['onu_group_range']
]
# Process in batches
aggregated_data: Dict[int, Dict[int, List[Dict[str, Optional[str]]]]] = {}
for i in range(0, len(all_combinations), CONFIG['batch_size']):
batch = all_combinations[i:i + CONFIG['batch_size']]
logger.info(f"Processing batch {i//CONFIG['batch_size'] + 1}")
results = await process_batch(scraper, batch)
for result in results:
if isinstance(result, Exception):
logger.error(f"Batch task failed with exception: {result}")
continue
for pon, onu_data in result.items():
if pon not in aggregated_data:
aggregated_data[pon] = {}
aggregated_data[pon].update(onu_data)
return aggregated_data
async def handle_data(request):
"""
HTTP handler for the configured endpoint.
Triggers data collection and streams the aggregated JSON data incrementally using SSE.
"""
start_time = datetime.now()
logger.info(f"Received {CONFIG['endpoint']} request at {start_time}")
# Initialize the SSE response
resp = web.StreamResponse(
status=200,
reason='OK',
headers={
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
'Connection': 'keep-alive',
}
)
await resp.prepare(request)
try:
async with AsyncScraper(CONFIG['base_url']) as scraper:
if not await scraper.login(CONFIG['username'], CONFIG['password']):
error_data = json.dumps({"error": "Login failed. Please check credentials."})
await resp.write(f"event: error\ndata: {error_data}\n\n".encode('utf-8'))
await resp.write_eof()
return resp
# Create all combinations of PON and ONU groups
all_combinations = [
(pon, onu_group)
for pon in CONFIG['pon_range']
for onu_group in CONFIG['onu_group_range']
]
aggregated_data: Dict[int, Dict[int, List[Dict[str, Optional[str]]]]] = {}
# Process in batches
for i in range(0, len(all_combinations), CONFIG['batch_size']):
batch = all_combinations[i:i + CONFIG['batch_size']]
logger.info(f"Processing batch {i // CONFIG['batch_size'] + 1}")
results = await process_batch(scraper, batch)
batch_data = {}
for result in results:
if isinstance(result, Exception):
logger.error(f"Batch task failed with exception: {result}")
continue
for pon, onu_data in result.items():
if pon not in aggregated_data:
aggregated_data[pon] = {}
aggregated_data[pon].update(onu_data)
if pon not in batch_data:
batch_data[pon] = {}
batch_data[pon].update(onu_data)
# Prepare the batch data to send
json_part = json.dumps(batch_data)
# Send the batch data as an SSE event
sse_event = f"event: batch\ndata: {json_part}\n\n"
await resp.write(sse_event.encode('utf-8'))
# Removed the deprecated drain call
end_time = datetime.now()
duration = end_time - start_time
logger.info(f"Data collection complete. Duration: {duration}")
# Send a final completion event
final_event = f"event: complete\ndata: {{\"status\": \"complete\", \"duration\": \"{duration}\"}}\n\n"
await resp.write(final_event.encode('utf-8'))
await resp.write_eof()
except Exception as e:
logger.error(f"Error in handle_data: {e}")
error_event = f"event: error\ndata: {{\"error\": \"{str(e)}\"}}\n\n"
await resp.write(error_event.encode('utf-8'))
await resp.write_eof()
return resp
async def init_app():
"""
Initializes the aiohttp web application.
"""
app = web.Application()
app.router.add_get(CONFIG['endpoint'], handle_data) # Use configurable endpoint
return app
def parse_args():
"""
Parses command-line arguments.
"""
parser = argparse.ArgumentParser(description="Async Scraper Server")
parser.add_argument(
'--ignore-row-length-warnings',
action='store_true',
help='Ignore warnings about row length mismatches in HTML tables.'
)
# You can add more CLI parameters here if needed.
return parser.parse_args()
def main():
"""
Main entry point to run the web server.
"""
# Parse CLI arguments
args = parse_args()
# Update CONFIG based on CLI arguments
CONFIG['ignore_row_length_warning'] = args.ignore_row_length_warnings
# Retrieve all IP addresses
local_ips = get_all_ip_addresses()
if not local_ips:
logger.warning("No non-loopback IPv4 addresses found. Falling back to 127.0.0.1")
local_ips = [CONFIG['host']]
endpoint = CONFIG.get('endpoint', '/data')
logger.info(f"Starting server on port {CONFIG['port']}")
logger.info("Accessible at the following URLs:")
for ip in local_ips:
logger.info(f"http://{ip}:{CONFIG['port']}{endpoint}")
loop = asyncio.get_event_loop()
app = loop.run_until_complete(init_app())
runner = web.AppRunner(app)
loop.run_until_complete(runner.setup())
site = web.TCPSite(runner, host=CONFIG['host'], port=CONFIG['port'])
loop.run_until_complete(site.start())
logger.info("Server is now running")
try:
loop.run_forever()
except KeyboardInterrupt:
logger.info("Shutting down server...")
finally:
loop.run_until_complete(runner.cleanup())
if name == "main":
main()
ASSISTANT
To implement a **graceful shutdown** in your asyncio-based `aiohttp` web server, you'll need to handle termination signals (such as `SIGINT` and `SIGTERM`) properly. This ensures that your server stops accepting new requests, completes any ongoing tasks, and cleans up resources before exiting.
Below is the modified version of your script with graceful shutdown implemented. The key changes include:
1. **Signal Handling**: Capturing termination signals to initiate the shutdown process.
2. **Refactoring `main()`**: Converting `main()` into an asynchronous function and using `asyncio.run()` for better management.
3. **Shutdown Procedures**: Ensuring the server stops accepting new connections and existing connections are properly closed.
### Updated Script with Graceful Shutdown
```python
import asyncio
import aiohttp
from aiohttp import web
from bs4 import BeautifulSoup
import json
from typing import List, Dict, Optional
import ssl
import logging
from datetime import datetime
import colorlog
import socket
import netifaces # Newly added
import argparse # Newly added
import sys # Newly added
import signal
# Configuration settings
CONFIG = {
'base_url': 'https://10.0.0.2',
'username': 'misael_campos',
'password': 'd337O8AetR3p4',
'post_endpoint': '/action/pononuopticalinfo.html',
'who': 100,
'onuid': 0,
'pon_range': range(1, 17),
'onu_group_range': range(0, 2),
'request_timeout': aiohttp.ClientTimeout(
total=240, # Total operation timeout
connect=120, # Connection timeout
sock_read=120 # Socket read timeout
),
'max_concurrent_requests': 8, # Reduced to prevent server overload
'max_retries': 3, # Maximum number of retry attempts
'retry_delay': 1, # Delay between retries in seconds
'batch_size': 8, # Number of concurrent requests per batch
'host': '0.0.0.0', # Host address
'port': 8080, # Port number
'endpoint': '/data', # Added endpoint path
'ignore_row_length_warning': False # New configuration option
}
# Configure logging
handler = colorlog.StreamHandler()
handler.setFormatter(colorlog.ColoredFormatter(
'%(log_color)s%(asctime)s [%(levelname)s] %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
log_colors={
'DEBUG': 'light_cyan',
'INFO': 'light_green',
'WARNING': 'light_yellow',
'ERROR': 'light_red',
'CRITICAL': 'light_red,bg_white',
}
))
logger = colorlog.getLogger(__name__)
logger.addHandler(handler)
logger.setLevel(logging.INFO)
# Remove any existing handlers to avoid duplicate logs
for hdlr in logger.handlers[:-1]:
logger.removeHandler(hdlr)
# Function to retrieve all IPv4 addresses
def get_all_ip_addresses() -> List[str]:
ips = []
interfaces = netifaces.interfaces()
for iface in interfaces:
addrs = netifaces.ifaddresses(iface)
ipv4 = addrs.get(netifaces.AF_INET)
if ipv4:
for addr in ipv4:
ip = addr.get('addr')
if ip and ip != '127.0.0.1':
ips.append(ip)
return ips
class AsyncScraper:
def __init__(self, base_url: str):
self.base_url = base_url.rstrip('/')
self.ssl_context = ssl.create_default_context()
self.ssl_context.check_hostname = False
self.ssl_context.verify_mode = ssl.CERT_NONE
self.session = None
self.sem = asyncio.Semaphore(CONFIG['max_concurrent_requests'])
async def __aenter__(self):
timeout = CONFIG['request_timeout']
self.session = aiohttp.ClientSession(
connector=aiohttp.TCPConnector(
ssl=self.ssl_context,
limit=CONFIG['max_concurrent_requests']
),
timeout=timeout,
headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'}
)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
await self.session.close()
async def login(self, username: str, password: str) -> bool:
login_url = f"{self.base_url}/action/main.html"
login_data = {
'user': username,
'pass': password,
'button': 'Login',
'who': '100',
}
for attempt in range(CONFIG['max_retries']):
try:
async with self.session.post(login_url, data=login_data) as response:
text = await response.text()
if "Sorry, you do not have access" in text:
logger.error("Login failed: Access denied")
return False
elif "Error" in text or "Incorrect password" in text:
logger.error("Login failed: Incorrect credentials")
return False
logger.info("Login successful!")
return True
except Exception as e:
logger.error(f"Login attempt {attempt + 1} failed: {e}")
if attempt < CONFIG['max_retries'] - 1:
await asyncio.sleep(CONFIG['retry_delay'])
else:
return False
async def fetch_pon_onu_optical_info(self, pon: int, onu_group: int) -> Optional[str]:
async with self.sem:
post_url = f"{self.base_url}{CONFIG['post_endpoint']}"
payload = {
'pon': pon,
'onu_group': onu_group,
'who': CONFIG['who'],
'onuid': CONFIG['onuid']
}
for attempt in range(CONFIG['max_retries']):
try:
async with self.session.post(post_url, data=payload) as response:
if response.status == 200:
content = await response.text()
logger.info(f"Successfully fetched data for PON {pon}, ONU Group {onu_group} (Attempt {attempt + 1})")
return content
else:
logger.warning(f"Error {response.status} for PON {pon}, ONU Group {onu_group} (Attempt {attempt + 1})")
except asyncio.TimeoutError:
logger.warning(f"Timeout fetching PON {pon}, ONU Group {onu_group} (Attempt {attempt + 1})")
except Exception as e:
logger.warning(f"Error fetching PON {pon}, ONU Group {onu_group} (Attempt {attempt + 1}): {e}")
if attempt < CONFIG['max_retries'] - 1:
await asyncio.sleep(CONFIG['retry_delay'])
logger.info(f"Retrying PON {pon}, ONU Group {onu_group}...")
else:
logger.error(f"All attempts failed for PON {pon}, ONU Group {onu_group}")
return None
def parse_html_table(self, html_content: str) -> List[Dict[str, Optional[str]]]:
if not html_content:
return []
soup = BeautifulSoup(html_content, 'html.parser')
table = soup.find('table', border="1", cellpadding="4", cellspacing="0")
if not table:
logger.warning("No table found in the HTML content")
return []
headers = []
header_row = table.find('tr')
if not header_row:
logger.warning("No header row found in the table")
return []
for th in header_row.find_all(['td', 'th']):
header_text = th.get_text(strip=True)
headers.append(header_text if header_text else f"Column_{len(headers)+1}")
data = []
for row in table.find_all('tr')[1:]:
cells = row.find_all('td')
if len(cells) != len(headers):
if not CONFIG['ignore_row_length_warning']:
logger.warning("Row does not match header length. Skipping row")
continue
row_data = {
header: cell.get_text(strip=True) or None
for header, cell in zip(headers, cells)
}
data.append(row_data)
return data
async def fetch_and_parse(self, pon: int, onu_group: int) -> Dict[int, Dict[int, List[Dict[str, Optional[str]]]]]:
html_content = await self.fetch_pon_onu_optical_info(pon, onu_group)
parsed_data = self.parse_html_table(html_content) if html_content else []
if not parsed_data:
logger.warning(f"No data parsed for PON {pon}, ONU Group {onu_group}")
return {pon: {onu_group: parsed_data}}
async def process_batch(self, batch: List[tuple]) -> List[Dict]:
tasks = [self.fetch_and_parse(pon, onu_group) for pon, onu_group in batch]
return await asyncio.gather(*tasks, return_exceptions=True)
async def collect_data(self) -> Dict[int, Dict[int, List[Dict[str, Optional[str]]]]]:
# Create all combinations of PON and ONU groups
all_combinations = [
(pon, onu_group)
for pon in CONFIG['pon_range']
for onu_group in CONFIG['onu_group_range']
]
# Process in batches
aggregated_data: Dict[int, Dict[int, List[Dict[str, Optional[str]]]]] = {}
for i in range(0, len(all_combinations), CONFIG['batch_size']):
batch = all_combinations[i:i + CONFIG['batch_size']]
logger.info(f"Processing batch {i//CONFIG['batch_size'] + 1}")
results = await self.process_batch(batch)
for result in results:
if isinstance(result, Exception):
logger.error(f"Batch task failed with exception: {result}")
continue
for pon, onu_data in result.items():
if pon not in aggregated_data:
aggregated_data[pon] = {}
aggregated_data[pon].update(onu_data)
return aggregated_data
async def handle_data(request):
"""
HTTP handler for the configured endpoint.
Triggers data collection and streams the aggregated JSON data incrementally using SSE.
"""
start_time = datetime.now()
logger.info(f"Received {CONFIG['endpoint']} request at {start_time}")
# Initialize the SSE response
resp = web.StreamResponse(
status=200,
reason='OK',
headers={
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
'Connection': 'keep-alive',
}
)
await resp.prepare(request)
try:
async with AsyncScraper(CONFIG['base_url']) as scraper:
if not await scraper.login(CONFIG['username'], CONFIG['password']):
error_data = json.dumps({"error": "Login failed. Please check credentials."})
await resp.write(f"event: error\ndata: {error_data}\n\n".encode('utf-8'))
await resp.write_eof()
return resp
# Collect data
aggregated_data = await scraper.collect_data()
for pon, onu_groups in aggregated_data.items():
for onu_group, data in onu_groups.items():
batch_data = {pon: {onu_group: data}}
json_part = json.dumps(batch_data)
sse_event = f"event: batch\ndata: {json_part}\n\n"
await resp.write(sse_event.encode('utf-8'))
await resp.drain() # Ensure data is sent
end_time = datetime.now()
duration = end_time - start_time
logger.info(f"Data collection complete. Duration: {duration}")
# Send a final completion event
final_event = f"event: complete\ndata: {{\"status\": \"complete\", \"duration\": \"{duration}\"}}\n\n"
await resp.write(final_event.encode('utf-8'))
await resp.write_eof()
except asyncio.CancelledError:
logger.info("Request handler cancelled.")
except Exception as e:
logger.error(f"Error in handle_data: {e}")
error_event = f"event: error\ndata: {{\"error\": \"{str(e)}\"}}\n\n"
await resp.write(error_event.encode('utf-8'))
await resp.write_eof()
return resp
async def init_app():
"""
Initializes the aiohttp web application.
"""
app = web.Application()
app.router.add_get(CONFIG['endpoint'], handle_data) # Use configurable endpoint
return app
def parse_args():
"""
Parses command-line arguments.
"""
parser = argparse.ArgumentParser(description="Async Scraper Server")
parser.add_argument(
'--ignore-row-length-warnings',
action='store_true',
help='Ignore warnings about row length mismatches in HTML tables.'
)
# You can add more CLI parameters here if needed.
return parser.parse_args()
async def shutdown(signal, loop, runner):
"""
Cleanup tasks tied to the service's shutdown.
"""
logger.info(f"Received exit signal {signal.name}...")
logger.info("Shutting down server gracefully...")
# Stop accepting new connections
await runner.cleanup()
# Wait a bit for existing connections to finish
tasks = [t for t in asyncio.all_tasks(loop) if t is not asyncio.current_task()]
list(map(lambda task: task.cancel(), tasks))
logger.info("Cancelling outstanding tasks")
await asyncio.gather(*tasks, return_exceptions=True)
loop.stop()
async def main():
"""
Main entry point to run the web server with graceful shutdown.
"""
# Parse CLI arguments
args = parse_args()
# Update CONFIG based on CLI arguments
CONFIG['ignore_row_length_warning'] = args.ignore_row_length_warnings
# Retrieve all IP addresses
local_ips = get_all_ip_addresses()
if not local_ips:
logger.warning("No non-loopback IPv4 addresses found. Falling back to 127.0.0.1")
local_ips = [CONFIG['host']]
endpoint = CONFIG.get('endpoint', '/data')
logger.info(f"Starting server on port {CONFIG['port']}")
logger.info("Accessible at the following URLs:")
for ip in local_ips:
logger.info(f"http://{ip}:{CONFIG['port']}{endpoint}")
app = await init_app()
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, host=CONFIG['host'], port=CONFIG['port'])
await site.start()
logger.info("Server is now running")
# Get the current event loop
loop = asyncio.get_running_loop()
# Create a shutdown event
shutdown_event = asyncio.Event()
# Define shutdown signals
signals = (signal.SIGINT, signal.SIGTERM)
for s in signals:
loop.add_signal_handler(
s, lambda s=s: asyncio.create_task(shutdown(s, loop, runner))
)
# Run until shutdown event is set
try:
await shutdown_event.wait()
except asyncio.CancelledError:
pass
if __name__ == "__main__":
try:
asyncio.run(main())
except (KeyboardInterrupt, SystemExit):
logger.info("Server shutdown complete.")
```
### Key Changes Explained
1. **Signal Handling Setup**:
- **Importing `signal`**: Added the `signal` module import to handle OS signals.
- **Defining `shutdown()` Coroutine**: This coroutine handles the cleanup process when a shutdown signal is received. It stops accepting new connections, cancels outstanding tasks, and stops the loop.
- **Registering Signal Handlers**: In the `main()` coroutine, signal handlers for `SIGINT` and `SIGTERM` are registered to trigger the `shutdown()` coroutine.
2. **Refactoring `main()`**:
- Converted `main()` to an `async` function to leverage `asyncio.run()` for managing the event loop.
- Used `asyncio.run(main())` to execute the `main()` coroutine, which is the recommended way to start asyncio programs in Python 3.7+.
3. **Graceful Shutdown Flow**:
- **Receiving a Signal**: When a shutdown signal is received (e.g., pressing `Ctrl+C`), the `shutdown()` coroutine is invoked.
- **Stopping the Server**: The `runner.cleanup()` method stops the server from accepting new connections and starts the cleanup process.
- **Cancelling Outstanding Tasks**: All ongoing tasks except the current one are cancelled to ensure the server shuts down promptly.
- **Stopping the Event Loop**: Finally, the event loop is stopped gracefully.
4. **Handling `CancelledError`**:
- In the `handle_data` handler, an `asyncio.CancelledError` is caught to handle cases where the handler is cancelled during shutdown.
5. **Using `asyncio.Event` for Shutdown Coordination**:
- An `asyncio.Event` named `shutdown_event` is used to keep the server running until a shutdown signal is received.
6. **Logging Enhancements**:
- Added more logging statements to provide better insights during the shutdown process.
### Additional Recommendations
- **Timeouts During Shutdown**: Depending on your application's needs, you might want to add timeouts to how long the server waits for ongoing tasks to complete before forcefully cancelling them.
- **Error Handling**: Ensure that all possible exceptions are appropriately handled, especially during the shutdown process, to avoid unexpected behavior.
- **Testing**: After implementing graceful shutdown, thoroughly test the server by initiating shutdowns during various stages of request handling to ensure that it behaves as expected.
### Conclusion
With these changes, your `aiohttp` server will gracefully handle shutdown signals, ensuring that ongoing requests are completed and resources are properly released before the application exits. This makes your server more robust and production-ready.