🚀 Day 28: Dive into the fascinating world of Asynchronous Programming with `asyncio`! 🐢 Tired of your program waiting? `asyncio` lets your Python code perform multiple tasks *seemingly* at once, especially great for I/O-bound operations like network requests or file operations. Say goodbye to blocking and hello to efficiency! 💨 Let's make your code run smarter, not harder.
```python
import asyncio
import time
async def perform_task(name, delay):
"""Simulates an I/O-bound operation."""
print(f"[{name}] Starting...")
await asyncio.sleep(delay) # Yield control back to the event loop
print(f"[{name}] Finished after {delay} seconds!")
return f"{name} done"
async def main():
start_time = time.time()
print("Main: Kicking off async tasks...")
# Run tasks concurrently
results = await asyncio.gather(
perform_task("Task A", 2),
perform_task("Task B", 1),
perform_task("Task C", 3)
)
end_time = time.time()
print(f"nMain: All tasks completed in {end_time - start_time:.2f} seconds.")
print(f"Main: Results: {results}")
if __name__ == "__main__":
asyncio.run(main())
```
Output example:
Main: Kicking off async tasks...
[Task A] Starting...
[Task B] Starting...
[Task C] Starting...
[Task B] Finished after 1 seconds!
[Task A] Finished after 2 seconds!
[Task C] Finished after 3 seconds!
Main: All tasks completed in 3.00 seconds.
Main: Results: ['Task A done', 'Task B done', 'Task C done']
💡 **Small Challenge:** Add another `async` function that attempts to read a (non-existent) file using `asyncio.to_thread` to simulate a blocking file read. How would you integrate it into the `asyncio.gather` call? What's the benefit of `asyncio.to_thread` here?
---
```python
import asyncio
import time
async def perform_task(name, delay):
"""Simulates an I/O-bound operation."""
print(f"[{name}] Starting...")
await asyncio.sleep(delay) # Yield control back to the event loop
print(f"[{name}] Finished after {delay} seconds!")
return f"{name} done"
async def main():
start_time = time.time()
print("Main: Kicking off async tasks...")
# Run tasks concurrently
results = await asyncio.gather(
perform_task("Task A", 2),
perform_task("Task B", 1),
perform_task("Task C", 3)
)
end_time = time.time()
print(f"nMain: All tasks completed in {end_time - start_time:.2f} seconds.")
print(f"Main: Results: {results}")
if __name__ == "__main__":
asyncio.run(main())
```
Output example:
Main: Kicking off async tasks...
[Task A] Starting...
[Task B] Starting...
[Task C] Starting...
[Task B] Finished after 1 seconds!
[Task A] Finished after 2 seconds!
[Task C] Finished after 3 seconds!
Main: All tasks completed in 3.00 seconds.
Main: Results: ['Task A done', 'Task B done', 'Task C done']
💡 **Small Challenge:** Add another `async` function that attempts to read a (non-existent) file using `asyncio.to_thread` to simulate a blocking file read. How would you integrate it into the `asyncio.gather` call? What's the benefit of `asyncio.to_thread` here?
---
Here's your Telegram post for Day 29:
---
**Day 29: Basic Database Interaction (SQLite) 💾**
Ever wondered how applications store all that important information? Today, we dive into the world of databases with Python's built-in `sqlite3` module! 🚀 SQLite is a super lightweight, file-based database – perfect for learning and small projects. You'll learn to connect, create tables, insert data, and retrieve it. Let's get persistent! ✨
```python
import sqlite3
# 1. Connect to the database (or create it if it doesn't exist)
# This will create a file named 'my_items.db' in your script's directory.
conn = sqlite3.connect('my_items.db')
cursor = conn.cursor()
# 2. Create a table (IF NOT EXISTS prevents errors if you run this multiple times)
cursor.execute('''
CREATE TABLE IF NOT EXISTS products (
id INTEGER PRIMARY KEY,
item_name TEXT NOT NULL,
quantity INTEGER NOT NULL
)
''')
# 3. Insert some data into the table
# Using placeholders (?) is crucial to prevent SQL injection vulnerabilities.
cursor.execute("INSERT INTO products (item_name, quantity) VALUES (?, ?)", ("Laptop", 5))
cursor.execute("INSERT INTO products (item_name, quantity) VALUES (?, ?)", ("Mouse", 12))
cursor.execute("INSERT INTO products (item_name, quantity) VALUES (?, ?)", ("Keyboard", 8))
cursor.execute("INSERT INTO products (item_name, quantity) VALUES (?, ?)", ("Monitor", 3))
# Important: Commit the changes to save them permanently to the database file
conn.commit()
# 4. Retrieve and print all data from the 'products' table
print("All products in the database:")
cursor.execute("SELECT * FROM products")
items = cursor.fetchall() # Fetches all rows from the last executed query
for item in items:
print(item)
# 5. Close the database connection
conn.close()
```
Output example:
```
All products in the database:
(1, 'Laptop', 5)
(2, 'Mouse', 12)
(3, 'Keyboard', 8)
(4, 'Monitor', 3)
```
Challenge Time! 🧠 Can you modify the `SELECT` query in the code to retrieve *only* products with a `quantity` greater than 10? Share your solution! 👇
---
---
**Day 29: Basic Database Interaction (SQLite) 💾**
Ever wondered how applications store all that important information? Today, we dive into the world of databases with Python's built-in `sqlite3` module! 🚀 SQLite is a super lightweight, file-based database – perfect for learning and small projects. You'll learn to connect, create tables, insert data, and retrieve it. Let's get persistent! ✨
```python
import sqlite3
# 1. Connect to the database (or create it if it doesn't exist)
# This will create a file named 'my_items.db' in your script's directory.
conn = sqlite3.connect('my_items.db')
cursor = conn.cursor()
# 2. Create a table (IF NOT EXISTS prevents errors if you run this multiple times)
cursor.execute('''
CREATE TABLE IF NOT EXISTS products (
id INTEGER PRIMARY KEY,
item_name TEXT NOT NULL,
quantity INTEGER NOT NULL
)
''')
# 3. Insert some data into the table
# Using placeholders (?) is crucial to prevent SQL injection vulnerabilities.
cursor.execute("INSERT INTO products (item_name, quantity) VALUES (?, ?)", ("Laptop", 5))
cursor.execute("INSERT INTO products (item_name, quantity) VALUES (?, ?)", ("Mouse", 12))
cursor.execute("INSERT INTO products (item_name, quantity) VALUES (?, ?)", ("Keyboard", 8))
cursor.execute("INSERT INTO products (item_name, quantity) VALUES (?, ?)", ("Monitor", 3))
# Important: Commit the changes to save them permanently to the database file
conn.commit()
# 4. Retrieve and print all data from the 'products' table
print("All products in the database:")
cursor.execute("SELECT * FROM products")
items = cursor.fetchall() # Fetches all rows from the last executed query
for item in items:
print(item)
# 5. Close the database connection
conn.close()
```
Output example:
```
All products in the database:
(1, 'Laptop', 5)
(2, 'Mouse', 12)
(3, 'Keyboard', 8)
(4, 'Monitor', 3)
```
Challenge Time! 🧠 Can you modify the `SELECT` query in the code to retrieve *only* products with a `quantity` greater than 10? Share your solution! 👇
---
In Python's `sqlite3` module, what method is essential to save changes (like `INSERT` or `UPDATE`) permanently to the database file after executing them?
Anonymous Quiz
0%
cursor.execute()
100%
conn.commit()
0%
0%
cursor.save_changes()
🎉 Wow, we've made it to Day 30! 🎉
Congratulations on completing a month of consistent Python learning! Today isn't just about new syntax, but about reflecting, consolidating, and dreaming big. It's time to start thinking about your *final project* – the culmination of everything you've learned. Let's get those creative juices flowing and brainstorm some ideas, then review what makes a good project. What will you build? 🤔💡
```python
import random
def generate_project_idea():
"""Generates a random Python project idea from a predefined list."""
ideas = [
"A simple To-Do List application (CLI or basic GUI)",
"A calculator program with basic operations",
"A password generator (with options for length, symbols, numbers)",
"A unit converter (e.g., Celsius to Fahrenheit, meters to feet)",
"A basic 'Guess the Number' game",
"A text-based adventure game",
"A file organizer (e.g., sort files by type)",
"A simple web scraper for product prices on a specific site",
"A weather app (using a public API like OpenWeatherMap)",
"A basic contact book with add/view/delete functionality"
]
return random.choice(ideas)
print("🚀 Your potential project idea for today is:")
print(f"👉 {generate_project_idea()}")
```
Output example:
🚀 Your potential project idea for today is:
👉 A password generator (with options for length, symbols, numbers)
*(Note: Your actual output will vary each time you run the code!)*
**Small Challenge:**
Pick one of the generated ideas (or come up with your own awesome project concept!) and list 3 core features you would implement first. How would you start planning the steps to build it? Share your thoughts! 👇
---
Congratulations on completing a month of consistent Python learning! Today isn't just about new syntax, but about reflecting, consolidating, and dreaming big. It's time to start thinking about your *final project* – the culmination of everything you've learned. Let's get those creative juices flowing and brainstorm some ideas, then review what makes a good project. What will you build? 🤔💡
```python
import random
def generate_project_idea():
"""Generates a random Python project idea from a predefined list."""
ideas = [
"A simple To-Do List application (CLI or basic GUI)",
"A calculator program with basic operations",
"A password generator (with options for length, symbols, numbers)",
"A unit converter (e.g., Celsius to Fahrenheit, meters to feet)",
"A basic 'Guess the Number' game",
"A text-based adventure game",
"A file organizer (e.g., sort files by type)",
"A simple web scraper for product prices on a specific site",
"A weather app (using a public API like OpenWeatherMap)",
"A basic contact book with add/view/delete functionality"
]
return random.choice(ideas)
print("🚀 Your potential project idea for today is:")
print(f"👉 {generate_project_idea()}")
```
Output example:
🚀 Your potential project idea for today is:
👉 A password generator (with options for length, symbols, numbers)
*(Note: Your actual output will vary each time you run the code!)*
**Small Challenge:**
Pick one of the generated ideas (or come up with your own awesome project concept!) and list 3 core features you would implement first. How would you start planning the steps to build it? Share your thoughts! 👇
---
When starting a new Python project, what is generally considered the *most crucial first step* after deciding on an idea?
Anonymous Quiz
100%
Writing all the code for the main functionality immediately.
0%
Designing the project architecture and breaking it into smaller tasks.
0%
Creating an exhaustive list of all possible advanced features.
0%
Picking a random Python library to integrate, regardless of need.
What is the primary purpose of `asyncio.loop.run_in_executor()` when integrating with blocking C functions via FFI?
Anonymous Quiz
0%
To move blocking C calls off the main event loop thread to prevent it from freezing.
0%
To convert a synchronous C function directly into a coroutine without using threads.
0%
To execute C code directly on the Python interpreter's main thread for maximum speed.
0%
To automatically compile and link C code at runtime within the event loop.
🌟 Day 32: Unlock the ultimate power of Python! 🌟 Ever wondered how frameworks create magical APIs or enforce specific class structures? Today, we dive deep into the #MetaclassMultiverse! 🤯 We'll explore how to forge custom type systems, going far beyond simple `type` calls to control class creation itself. Get ready to redefine what a Python class can be! Let's build some powerful magic together. ✨ #PythonLearning #AdvancedPython #Metaclasses
Metaclasses allow you to intercept class creation and inject behavior, attributes, or methods before the class is even fully defined. This enables powerful patterns for validation, registration, and framework building!
```python
import datetime
# 🚀 Our custom metaclass
class MyMeta(type):
"""
A metaclass that automatically adds a creation timestamp and a 'greet' method
to any class that uses it.
"""
def __new__(cls, name, bases, dct):
# 1. Add a creation timestamp to the class dictionary
dct['_creation_timestamp'] = datetime.datetime.now()
# 2. Add a default 'greet' method if not already defined
if 'greet' not in dct:
def greet_method(self):
return (f"Hello from {self.__class__.__name__}! "
f"Created at {self._creation_timestamp.strftime('%Y-%m-%d %H:%M:%S')}")
dct['greet'] = greet_method
# 3. Call the base type's __new__ to complete class creation
return super().__new__(cls, name, bases, dct)
# 🧙♀️ Classes using our custom metaclass
class Character(metaclass=MyMeta):
def __init__(self, name):
self.name = name
class Item(metaclass=MyMeta):
def __init__(self, name):
self.name = name
# This class defines its own greet, which will override the metaclass's
def greet(self):
return f"Hello, I am {self.name}, an item created magically!"
# ✨ Let's see them in action!
hero = Character("Alice")
sword = Item("Excalibur")
print(hero.greet())
print(sword.greet()) # Notice how Item's custom greet takes precedence!
print(f"Character created at: {Character._creation_timestamp}")
```
Output example:
```
Hello from Character! Created at 2023-10-27 10:30:00 # (Timestamp will vary)
Hello, I am Excalibur, an item created magically!
Character created at: 2023-10-27 10:30:00 # (Timestamp will vary)
```
💡 **Your Challenge:** Modify `MyMeta` to automatically add a class-level `_registry` dictionary. For every class that uses `MyMeta`, register the class itself in this `_registry` using its `name` as the key. How would you then access this registry from another part of your code to list all registered classes?
---
Metaclasses allow you to intercept class creation and inject behavior, attributes, or methods before the class is even fully defined. This enables powerful patterns for validation, registration, and framework building!
```python
import datetime
# 🚀 Our custom metaclass
class MyMeta(type):
"""
A metaclass that automatically adds a creation timestamp and a 'greet' method
to any class that uses it.
"""
def __new__(cls, name, bases, dct):
# 1. Add a creation timestamp to the class dictionary
dct['_creation_timestamp'] = datetime.datetime.now()
# 2. Add a default 'greet' method if not already defined
if 'greet' not in dct:
def greet_method(self):
return (f"Hello from {self.__class__.__name__}! "
f"Created at {self._creation_timestamp.strftime('%Y-%m-%d %H:%M:%S')}")
dct['greet'] = greet_method
# 3. Call the base type's __new__ to complete class creation
return super().__new__(cls, name, bases, dct)
# 🧙♀️ Classes using our custom metaclass
class Character(metaclass=MyMeta):
def __init__(self, name):
self.name = name
class Item(metaclass=MyMeta):
def __init__(self, name):
self.name = name
# This class defines its own greet, which will override the metaclass's
def greet(self):
return f"Hello, I am {self.name}, an item created magically!"
# ✨ Let's see them in action!
hero = Character("Alice")
sword = Item("Excalibur")
print(hero.greet())
print(sword.greet()) # Notice how Item's custom greet takes precedence!
print(f"Character created at: {Character._creation_timestamp}")
```
Output example:
```
Hello from Character! Created at 2023-10-27 10:30:00 # (Timestamp will vary)
Hello, I am Excalibur, an item created magically!
Character created at: 2023-10-27 10:30:00 # (Timestamp will vary)
```
💡 **Your Challenge:** Modify `MyMeta` to automatically add a class-level `_registry` dictionary. For every class that uses `MyMeta`, register the class itself in this `_registry` using its `name` as the key. How would you then access this registry from another part of your code to list all registered classes?
---
What is the primary purpose of a Python metaclass?
Anonymous Quiz
100%
To define the methods and attributes of an instance of a class.
0%
To control the creation of classes, allowing modification before the class is fully formed.
0%
To enable multiple inheritance in a more structured way.
0%
To provide a static type checking mechanism during development.
Which of the following special methods in Python is invoked when an attribute is accessed on an object, *regardless* of whether the attribute exists on the object or its class, making it ideal for comprehensive proxying?
Anonymous Quiz
100%
A) __getattr__
0%
B) __getattribute__
0%
C) __call__
0%
D) __dir__
🎉 Day 34: Unlocking Python's Object Model! 🐍 Ever wondered how `@property` works its magic, or why methods behave like they do? Today, we dive into the **Descriptor Protocol**! 🕵️♂️ This powerful mechanism allows you to customize attribute access (`__get__`, `__set__`, `__delete__`) for class variables. Master descriptors, and you'll gain unparalleled control over your objects' behavior, laying the groundwork for elegant and robust designs. Ready to peek behind the curtain? ✨
```python
class LoggingDescriptor:
"""
A simple descriptor that logs attribute access.
"""
def __set_name__(self, owner, name):
# Called when the descriptor is assigned to a class attribute
self.public_name = name
self.private_name = '_' + name # Store value in a private attribute
def __get__(self, obj, obj_type=None):
# Called when the attribute is accessed (e.g., obj.value)
if obj is None:
return self # Accessing descriptor directly from class (MyClass.value)
print(f"Accessing '{self.public_name}' from {obj_type.__name__}")
return getattr(obj, self.private_name, None)
def __set__(self, obj, value):
# Called when the attribute is assigned (e.g., obj.value = 10)
print(f"Setting '{self.public_name}' to {value} in {obj.__class__.__name__}")
setattr(obj, self.private_name, value)
def __delete__(self, obj):
# Called when the attribute is deleted (e.g., del obj.value)
print(f"Deleting '{self.public_name}' from {obj.__class__.__name__}")
delattr(obj, self.private_name)
class MyClass:
value = LoggingDescriptor() # 'value' is an instance of our descriptor
another_value = LoggingDescriptor()
def __init__(self, initial_value):
self.value = initial_value # This assignment triggers LoggingDescriptor.__set__
# Demonstrate usage
print("--- Creating instance ---")
obj = MyClass(10)
print("n--- Accessing attributes ---")
print(f"obj.value: {obj.value}")
print(f"obj.another_value: {obj.another_value}") # Accessing an unset descriptor
print("n--- Modifying attribute ---")
obj.value = 20
print("n--- Accessing again ---")
print(f"obj.value: {obj.value}")
print("n--- Deleting attribute ---")
del obj.value
print("n--- Accessing after delete ---")
try:
print(obj.value)
except AttributeError as e:
print(f"Error accessing deleted attribute: {e}")
print("n--- Accessing descriptor directly from class ---")
print(MyClass.value)
```
Output example:
```
--- Creating instance ---
Setting 'value' to 10 in MyClass
--- Accessing attributes ---
Accessing 'value' from MyClass
obj.value: 10
Accessing 'another_value' from MyClass
obj.another_value: None
--- Modifying attribute ---
Setting 'value' to 20 in MyClass
--- Accessing again ---
Accessing 'value' from MyClass
obj.value: 20
--- Deleting attribute ---
Deleting 'value' from MyClass
--- Accessing after delete ---
Error accessing deleted attribute: 'MyClass' object has no attribute '_value'
--- Accessing descriptor directly from class ---
<__main__.LoggingDescriptor object at 0x...>
```
💡 **Your Challenge:** Modify the `LoggingDescriptor` to add **type validation**. For instance, ensure `value` can only be set to integers. If a non-integer is assigned, raise a `TypeError`! Can you make it even more generic to accept a type hint during descriptor instantiation (e.g., `value = ValidatedDescriptor(int)`)? 🚀
```python
class LoggingDescriptor:
"""
A simple descriptor that logs attribute access.
"""
def __set_name__(self, owner, name):
# Called when the descriptor is assigned to a class attribute
self.public_name = name
self.private_name = '_' + name # Store value in a private attribute
def __get__(self, obj, obj_type=None):
# Called when the attribute is accessed (e.g., obj.value)
if obj is None:
return self # Accessing descriptor directly from class (MyClass.value)
print(f"Accessing '{self.public_name}' from {obj_type.__name__}")
return getattr(obj, self.private_name, None)
def __set__(self, obj, value):
# Called when the attribute is assigned (e.g., obj.value = 10)
print(f"Setting '{self.public_name}' to {value} in {obj.__class__.__name__}")
setattr(obj, self.private_name, value)
def __delete__(self, obj):
# Called when the attribute is deleted (e.g., del obj.value)
print(f"Deleting '{self.public_name}' from {obj.__class__.__name__}")
delattr(obj, self.private_name)
class MyClass:
value = LoggingDescriptor() # 'value' is an instance of our descriptor
another_value = LoggingDescriptor()
def __init__(self, initial_value):
self.value = initial_value # This assignment triggers LoggingDescriptor.__set__
# Demonstrate usage
print("--- Creating instance ---")
obj = MyClass(10)
print("n--- Accessing attributes ---")
print(f"obj.value: {obj.value}")
print(f"obj.another_value: {obj.another_value}") # Accessing an unset descriptor
print("n--- Modifying attribute ---")
obj.value = 20
print("n--- Accessing again ---")
print(f"obj.value: {obj.value}")
print("n--- Deleting attribute ---")
del obj.value
print("n--- Accessing after delete ---")
try:
print(obj.value)
except AttributeError as e:
print(f"Error accessing deleted attribute: {e}")
print("n--- Accessing descriptor directly from class ---")
print(MyClass.value)
```
Output example:
```
--- Creating instance ---
Setting 'value' to 10 in MyClass
--- Accessing attributes ---
Accessing 'value' from MyClass
obj.value: 10
Accessing 'another_value' from MyClass
obj.another_value: None
--- Modifying attribute ---
Setting 'value' to 20 in MyClass
--- Accessing again ---
Accessing 'value' from MyClass
obj.value: 20
--- Deleting attribute ---
Deleting 'value' from MyClass
--- Accessing after delete ---
Error accessing deleted attribute: 'MyClass' object has no attribute '_value'
--- Accessing descriptor directly from class ---
<__main__.LoggingDescriptor object at 0x...>
```
💡 **Your Challenge:** Modify the `LoggingDescriptor` to add **type validation**. For instance, ensure `value` can only be set to integers. If a non-integer is assigned, raise a `TypeError`! Can you make it even more generic to accept a type hint during descriptor instantiation (e.g., `value = ValidatedDescriptor(int)`)? 🚀
Which of the following methods are specifically part of the Descriptor Protocol in Python, controlling attribute access?
Anonymous Quiz
0%
A) __call__
0%
B) __get__
0%
C) __init__
0%
D) __str__
Hey Pythonistas! 🐍 Get ready to unlock some serious processing power!
### Day 35: Coroutine Superpowers: Building High-Performance Data Pipelines with Asynchronous Generators 🚀
Beyond basic `asyncio`, today we dive into the magic of **Asynchronous Generators**. ✨ They combine the power of `async/await` for non-blocking operations with the memory efficiency of generators, making them perfect for building high-performance, back-pressured data pipelines. Imagine processing huge datasets or real-time streams without blocking your event loop! Let's see how to chain these powerful constructs for efficient data flow. ⚡️
```python
import asyncio
import random
async def produce_data(count: int, delay: float):
"""An asynchronous generator that yields simulated data."""
print("Producer started 📦")
for i in range(1, count + 1):
# Simulate async I/O or a network call
await asyncio.sleep(delay + random.uniform(0, 0.1))
print(f" Producer yielded: {i}")
yield i
print("Producer finished ✅")
async def process_data(data_stream):
"""An asynchronous generator that processes data from another async generator."""
print("Processor started ⚙️")
async for item in data_stream:
if item % 2 == 0: # Only process even numbers
processed_item = item * 10
print(f" Processor transformed: {item} -> {processed_item}")
yield processed_item
else:
print(f" Processor skipped: {item} (odd)")
print("Processor finished ✅")
async def consume_data(processed_stream):
"""An asynchronous function that consumes data from an async generator."""
print("Consumer started 📈")
total_processed = 0
async for item in processed_stream:
print(f" Consumer received: {item}")
total_processed += 1
print(f"Consumer finished. Total items received: {total_processed} ✅")
async def main():
print("--- Starting Pipeline ---")
data_source = produce_data(5, 0.2)
# Chain async generators: producer -> processor -> consumer
processed_pipeline = process_data(data_source)
await consume_data(processed_pipeline)
print("--- Pipeline Finished ---")
if __name__ == "__main__":
asyncio.run(main())
```
Output example:
```
--- Starting Pipeline ---
Consumer started 📈
Processor started ⚙️
Producer started 📦
Producer yielded: 1
Processor skipped: 1 (odd)
Producer yielded: 2
Processor transformed: 2 -> 20
Consumer received: 20
Producer yielded: 3
Processor skipped: 3 (odd)
Producer yielded: 4
Processor transformed: 4 -> 40
Consumer received: 40
Producer yielded: 5
Processor skipped: 5 (odd)
Producer finished ✅
Processor finished ✅
Consumer finished. Total items received: 2 ✅
--- Pipeline Finished ---
```
**Small Challenge:**
Add another asynchronous generator `log_data(processed_stream)` between `process_data` and `consume_data`. This new generator should simply print each item it receives before yielding it again, simulating logging or inspection without modification. Chain it into the pipeline! 🕵️♀️
---
### Day 35: Coroutine Superpowers: Building High-Performance Data Pipelines with Asynchronous Generators 🚀
Beyond basic `asyncio`, today we dive into the magic of **Asynchronous Generators**. ✨ They combine the power of `async/await` for non-blocking operations with the memory efficiency of generators, making them perfect for building high-performance, back-pressured data pipelines. Imagine processing huge datasets or real-time streams without blocking your event loop! Let's see how to chain these powerful constructs for efficient data flow. ⚡️
```python
import asyncio
import random
async def produce_data(count: int, delay: float):
"""An asynchronous generator that yields simulated data."""
print("Producer started 📦")
for i in range(1, count + 1):
# Simulate async I/O or a network call
await asyncio.sleep(delay + random.uniform(0, 0.1))
print(f" Producer yielded: {i}")
yield i
print("Producer finished ✅")
async def process_data(data_stream):
"""An asynchronous generator that processes data from another async generator."""
print("Processor started ⚙️")
async for item in data_stream:
if item % 2 == 0: # Only process even numbers
processed_item = item * 10
print(f" Processor transformed: {item} -> {processed_item}")
yield processed_item
else:
print(f" Processor skipped: {item} (odd)")
print("Processor finished ✅")
async def consume_data(processed_stream):
"""An asynchronous function that consumes data from an async generator."""
print("Consumer started 📈")
total_processed = 0
async for item in processed_stream:
print(f" Consumer received: {item}")
total_processed += 1
print(f"Consumer finished. Total items received: {total_processed} ✅")
async def main():
print("--- Starting Pipeline ---")
data_source = produce_data(5, 0.2)
# Chain async generators: producer -> processor -> consumer
processed_pipeline = process_data(data_source)
await consume_data(processed_pipeline)
print("--- Pipeline Finished ---")
if __name__ == "__main__":
asyncio.run(main())
```
Output example:
```
--- Starting Pipeline ---
Consumer started 📈
Processor started ⚙️
Producer started 📦
Producer yielded: 1
Processor skipped: 1 (odd)
Producer yielded: 2
Processor transformed: 2 -> 20
Consumer received: 20
Producer yielded: 3
Processor skipped: 3 (odd)
Producer yielded: 4
Processor transformed: 4 -> 40
Consumer received: 40
Producer yielded: 5
Processor skipped: 5 (odd)
Producer finished ✅
Processor finished ✅
Consumer finished. Total items received: 2 ✅
--- Pipeline Finished ---
```
**Small Challenge:**
Add another asynchronous generator `log_data(processed_stream)` between `process_data` and `consume_data`. This new generator should simply print each item it receives before yielding it again, simulating logging or inspection without modification. Chain it into the pipeline! 🕵️♀️
---
Here's your Telegram post for Day 36 of the Python learning series!
---
🔥🐍 **Day 36: Cracking the CPython Egg!** 🤯🚀
Ready to peel back the layers of Python and truly understand how it works under the hood? Today, we're not just coding; we're *cracking the CPython egg* and exploring the advanced mechanisms that power your applications. Get ready to peek into bytecode, manipulate runtime, and sculpt complex systems with meta-programming. This isn't your average Python – it's the stuff that builds robust frameworks! ⚙️🔍
Let's start by looking at Python's bytecode, the low-level instructions the CPython interpreter executes. The `dis` module is our secret weapon here!
```python
import dis
def calculate_and_decide(a, b):
"""
A simple function to demonstrate Python bytecode.
It calculates a sum and applies a conditional transformation.
"""
intermediate_result = a + b
if intermediate_result > 10:
return intermediate_result * 2
else:
return intermediate_result + 1
print("--- Disassembling 'calculate_and_decide' function ---")
dis.dis(calculate_and_decide)
```
Output example:
```
--- Disassembling 'calculate_and_decide' function ---
8 0 LOAD_FAST 0 (a)
2 LOAD_FAST 1 (b)
4 BINARY_ADD
6 STORE_FAST 2 (intermediate_result)
9 8 LOAD_FAST 2 (intermediate_result)
10 LOAD_CONST 1 (10)
12 COMPARE_OP 4 (>)
14 POP_JUMP_IF_FALSE 24
10 16 LOAD_FAST 2 (intermediate_result)
18 LOAD_CONST 2 (2)
20 BINARY_MULTIPLY
22 RETURN_VALUE
12 24 LOAD_FAST 2 (intermediate_result)
26 LOAD_CONST 3 (1)
28 BINARY_ADD
30 RETURN_VALUE
```
### Small Challenge:
Try disassembling a function that uses a `for` loop or a `while` loop. Can you identify the bytecode instructions responsible for the iteration, comparison, and jumping logic? Share your insights and which specific instructions you found interesting!
---
---
🔥🐍 **Day 36: Cracking the CPython Egg!** 🤯🚀
Ready to peel back the layers of Python and truly understand how it works under the hood? Today, we're not just coding; we're *cracking the CPython egg* and exploring the advanced mechanisms that power your applications. Get ready to peek into bytecode, manipulate runtime, and sculpt complex systems with meta-programming. This isn't your average Python – it's the stuff that builds robust frameworks! ⚙️🔍
Let's start by looking at Python's bytecode, the low-level instructions the CPython interpreter executes. The `dis` module is our secret weapon here!
```python
import dis
def calculate_and_decide(a, b):
"""
A simple function to demonstrate Python bytecode.
It calculates a sum and applies a conditional transformation.
"""
intermediate_result = a + b
if intermediate_result > 10:
return intermediate_result * 2
else:
return intermediate_result + 1
print("--- Disassembling 'calculate_and_decide' function ---")
dis.dis(calculate_and_decide)
```
Output example:
```
--- Disassembling 'calculate_and_decide' function ---
8 0 LOAD_FAST 0 (a)
2 LOAD_FAST 1 (b)
4 BINARY_ADD
6 STORE_FAST 2 (intermediate_result)
9 8 LOAD_FAST 2 (intermediate_result)
10 LOAD_CONST 1 (10)
12 COMPARE_OP 4 (>)
14 POP_JUMP_IF_FALSE 24
10 16 LOAD_FAST 2 (intermediate_result)
18 LOAD_CONST 2 (2)
20 BINARY_MULTIPLY
22 RETURN_VALUE
12 24 LOAD_FAST 2 (intermediate_result)
26 LOAD_CONST 3 (1)
28 BINARY_ADD
30 RETURN_VALUE
```
### Small Challenge:
Try disassembling a function that uses a `for` loop or a `while` loop. Can you identify the bytecode instructions responsible for the iteration, comparison, and jumping logic? Share your insights and which specific instructions you found interesting!
---
What is the primary purpose of the `dis` module in Python?
Anonymous Quiz
0%
A. To convert Python code into machine-readable assembly language.
0%
B. To disassemble Python bytecode, showing how the CPython interpreter executes code.
0%
C. To optimize Python code by suggesting performance improvements.
0%
D. To encrypt Python source code for security purposes.
🚀 Day 37 Alert! Today, we're diving deep into Python's performance secrets with **Zero-Cost Abstractions**. Ever wondered how to squeeze every byte and cycle out of your code? We'll uncover techniques like `__slots__` that dramatically reduce memory footprint by eliminating `__dict__`, bringing C-like efficiency to your Python objects. Get ready to sculpt leaner, faster applications! ⚡️
```python
import sys
# Class without __slots__
# Each instance has its own __dict__ for attributes
class RegularPoint:
def __init__(self, x, y):
self.x = x
self.y = y
# Class with __slots__
# Attributes are stored directly in the instance, no __dict__
class SlottedPoint:
__slots__ = ('x', 'y') # Explicitly declare attributes
def __init__(self, x, y):
self.x = x
self.y = y
# Demonstrate memory usage
p_regular = RegularPoint(10, 20)
p_slotted = SlottedPoint(10, 20)
print(f"Memory size of RegularPoint: {sys.getsizeof(p_regular)} bytes")
print(f"Memory size of SlottedPoint: {sys.getsizeof(p_slotted)} bytes")
# Characteristic: __slots__ prevents adding new attributes dynamically
try:
p_slotted.z = 30
except AttributeError as e:
print(f"nAttempted to add new attribute to SlottedPoint: {e}")
# Check for __dict__ presence
print(f"RegularPoint has __dict__: {'__dict__' in p_regular.__dir__()}")
print(f"SlottedPoint has __dict__: {'__dict__' in p_slotted.__dir__()}")
```
Output example:
```
Memory size of RegularPoint: 56 bytes
Memory size of SlottedPoint: 40 bytes
Attempted to add new attribute to SlottedPoint: 'SlottedPoint' object has no attribute 'z'
RegularPoint has __dict__: True
SlottedPoint has __dict__: False
```
💡 **Your Challenge:** How does `__slots__` impact inheritance? What happens if a subclass *doesn't* define `__slots__` but its parent does? Share your thoughts! 👇
---
```python
import sys
# Class without __slots__
# Each instance has its own __dict__ for attributes
class RegularPoint:
def __init__(self, x, y):
self.x = x
self.y = y
# Class with __slots__
# Attributes are stored directly in the instance, no __dict__
class SlottedPoint:
__slots__ = ('x', 'y') # Explicitly declare attributes
def __init__(self, x, y):
self.x = x
self.y = y
# Demonstrate memory usage
p_regular = RegularPoint(10, 20)
p_slotted = SlottedPoint(10, 20)
print(f"Memory size of RegularPoint: {sys.getsizeof(p_regular)} bytes")
print(f"Memory size of SlottedPoint: {sys.getsizeof(p_slotted)} bytes")
# Characteristic: __slots__ prevents adding new attributes dynamically
try:
p_slotted.z = 30
except AttributeError as e:
print(f"nAttempted to add new attribute to SlottedPoint: {e}")
# Check for __dict__ presence
print(f"RegularPoint has __dict__: {'__dict__' in p_regular.__dir__()}")
print(f"SlottedPoint has __dict__: {'__dict__' in p_slotted.__dir__()}")
```
Output example:
```
Memory size of RegularPoint: 56 bytes
Memory size of SlottedPoint: 40 bytes
Attempted to add new attribute to SlottedPoint: 'SlottedPoint' object has no attribute 'z'
RegularPoint has __dict__: True
SlottedPoint has __dict__: False
```
💡 **Your Challenge:** How does `__slots__` impact inheritance? What happens if a subclass *doesn't* define `__slots__` but its parent does? Share your thoughts! 👇
---
What is the primary benefit of using `__slots__` in a Python class?
Anonymous Quiz
0%
To significantly reduce the memory footprint of objects.
0%
To enforce static typing for class attributes.
0%
To prevent instances from being garbage collected.
0%
To enable direct access to C-level memory structures.