2tony2

from typing import TypedDict

class DataPipelineConfig(TypedDict):
    name: str
    batch_size: int
    is_active: bool

def display_pipeline_config(config: DataPipelineConfig) -> str:
    status = 'active' if config['is_active'] else 'inactive'
    return (f"Pipeline {config['name']} with batch size {config['batch_size']} is currently {status}.")

2tony2

from typing import Optional

def process_data(file_path: str, delimiter: Optional[str] = None) -> str:
    delimiter_info = f" with delimiter '{delimiter}'" if delimiter else ""
    return f"Processing file at {file_path}{delimiter_info}"

print(process_data("/path/to/file.csv"))  # Output: Processing file at /path/to/file.csv
print(process_data("/path/to/file.csv", ","))  # Output: Processing file at /path/to/file.csv with delimiter ','

2tony2

from typing import Literal

def set_environment(env: Literal["development", "staging", "production"]) -> str:
    return f"Environment set to {env}"

print(set_environment("development"))  # Output: Environment set to development
print(set_environment("production"))  # Output: Environment set to production

# The following would raise a type error with a type checker like mypy, as "test" is not a valid literal
# print(set_environment("test"))

2tony2

from enum import Enum

class Environment(Enum):
    DEVELOPMENT = "development"
    STAGING = "staging"
    PRODUCTION = "production"

def set_environment(env: Environment) -> str:
    return f"Environment set to {env.value}"

2tony2

class DataPipeline:
    def __init__(self, name: str, batch_size: int):
        self.name = name
        self.batch_size = batch_size
        self.status = "inactive"

pipeline = DataPipeline(name="ETL Pipeline", batch_size=500)
print(vars(pipeline))
# Output: {'name': 'ETL Pipeline', 'batch_size': 500, 'status': 'inactive'}

2tony2

print(dir(pipeline))
# Output: ['__class__', '__delattr__', '__dict__', '__dir__', '__doc__', '__eq__', '__format__', '__ge__', '__getattribute__', '__gt__', '__hash__', '__init__', '__init_subclass__', '__le__', '__lt__', '__module__', '__ne__', '__new__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__', '__str__', '__subclasshook__', '__weakref__', 'batch_size', 'name', 'status']

2tony2

print(type(pipeline))
# Output: <class '__main__.DataPipeline'>
print(type(pipeline.name))
# Output: <class 'str'>
print(type(pipeline.batch_size))
# Output: <class 'int'>

2tony2

import inspect

# Get the class definition of the object
print(inspect.getmembers(pipeline))
# Output: [('__class__', <class '__main__.DataPipeline'>), ('__delattr__', <method-wrapper '__delattr__' of DataPipeline object at 0x7f8b6f3b1f10>), ('__dict__', {'name': 'ETL Pipeline', 'batch_size': 500, 'status': 'inactive'}), ('__dir__', <built-in method __dir__ of DataPipeline object at 0x7f8b6f3b1f10>), ...]

# Get the signature of the __init__ method
print(inspect.signature(DataPipeline.__init__))
# Output: (self, name: str, batch_size: int)

2tony2

import pandas as pd

# Example of reading a CSV file in chunks with Pandas
chunk_size = 1000
for chunk in pd.read_csv('large_file.csv', chunksize=chunk_size):
    process(chunk)  # Replace with your processing logic

2tony2

my_list = [1, 2, 3]
for item in my_list:
    print(item)