Python OOP: Classes, __init__, self, Instance vs Class Variables, Methods, Properties, Dunder Methods, and Every Pattern Data Engineers Need

You have been writing functions. Functions take data in, transform it, and return results. That works brilliantly for scripts and one-off pipelines. But when your codebase grows — when you have 20 tables to load, each with different configs, different transformations, and different validation rules — functions alone start to feel like a bag of loose tools. You need a toolbox.

Object-Oriented Programming (OOP) is that toolbox. A class is a blueprint. An object is a toolbox built from that blueprint. Each toolbox (object) carries its own data (attributes) and its own set of tools (methods) that know how to work with that data.

Real-life analogy: Think of a class like an employee ID badge template at a company. The template defines the fields: name, department, employee ID, access level. Every new employee (object) gets a badge printed from that template, but each badge has different values — “Naveen, Engineering, E1001, Level 3.” The template is the class. Each badge is an instance. The template does not change when you hire a new person — you just print a new badge with their information.

Table of Contents

• Why OOP? When Functions Are Not Enough
• Your First Class
• Defining a Class
• Creating Objects (Instances)
• The __init__ Method (Constructor)
• What __init__ Does
• Default Parameters in __init__
• Validation in __init__
• self — The Current Instance
• What self Actually Is
• Why self Is Explicit in Python
• Attributes
• Instance Attributes vs Class Attributes
• When to Use Each
• The Mutable Class Attribute Trap
• Methods
• Instance Methods
• Class Methods (@classmethod)
• Static Methods (@staticmethod)
• Choosing Between Method Types
• Properties (@property)
• Why Properties
• Getter, Setter, Deleter
• Computed Properties
• Dunder (Magic) Methods
• __str__ and __repr__
• __len__
• __eq__ and __lt__ (Comparison)
• __getitem__ (Indexing)
• __enter__ and __exit__ (Context Manager)
• Encapsulation — Public, Protected, Private
• Naming Conventions
• Name Mangling
• Data Engineering Patterns
• Pipeline Configuration Class
• Table Loader Class
• Connection Manager (Context Manager)
• Data Validator Class
• Pipeline Result Tracker
• Common Mistakes
• Interview Questions
• Wrapping Up

Why OOP? When Functions Are Not Enough

# Without OOP — functions with scattered state
def load_table(table_name, source_conn, target_conn, schema, batch_size, retry_count):
    # 6 parameters passed around everywhere
    pass

def validate_table(table_name, rules, schema, null_threshold):
    pass

def log_result(table_name, row_count, duration, status, error_msg):
    pass

# Every function needs the same parameters. Add a new field?
# Change it in EVERY function signature. Forget one? Bug.

# With OOP — data and behavior live together
class TablePipeline:
    def __init__(self, table_name, config):
        self.table_name = table_name
        self.config = config
        self.row_count = 0
        self.status = "pending"

    def load(self):       # self carries all the state
        pass

    def validate(self):   # no need to pass table_name, config — they are on self
        pass

    def log_result(self):
        pass

# Clean, organized, everything in one place
customers = TablePipeline("customers", customer_config)
customers.load()
customers.validate()
customers.log_result()

Your First Class

Defining a Class

# A class is a blueprint — it defines what data and behavior objects will have

class Employee:
    """Represents an employee in the company."""

    def __init__(self, name, department, employee_id):
        # These are INSTANCE ATTRIBUTES — unique to each employee
        self.name = name
        self.department = department
        self.employee_id = employee_id
        self.is_active = True   # Default value

    def introduce(self):
        """Instance method — operates on this specific employee."""
        return f"Hi, I am {self.name} from {self.department} (ID: {self.employee_id})"

    def deactivate(self):
        """Change the employee's status."""
        self.is_active = False
        return f"{self.name} has been deactivated"

Creating Objects (Instances)

# Creating instances — each is a separate object with its own data
naveen = Employee("Naveen", "Engineering", "E1001")
alice = Employee("Alice", "Data Science", "E1002")
bob = Employee("Bob", "Analytics", "E1003")

# Each object has its OWN copy of the attributes
print(naveen.name)          # "Naveen"
print(alice.name)           # "Alice"
print(naveen.department)    # "Engineering"
print(alice.department)     # "Data Science"

# Calling methods
print(naveen.introduce())   # "Hi, I am Naveen from Engineering (ID: E1001)"
print(alice.introduce())    # "Hi, I am Alice from Data Science (ID: E1002)"

# Modifying one object does NOT affect others
naveen.deactivate()
print(naveen.is_active)     # False
print(alice.is_active)      # True — Alice is unaffected

# Check the type
print(type(naveen))          # <class '__main__.Employee'>
print(isinstance(naveen, Employee))  # True

The __init__ Method (Constructor)

What __init__ Does

# __init__ is called AUTOMATICALLY when you create an object
# It sets up the object's initial state

class DatabaseConnection:
    def __init__(self, host, port, database, username):
        # __init__ receives the arguments you pass to the class
        # self refers to the NEW object being created
        self.host = host
        self.port = port
        self.database = database
        self.username = username
        self.is_connected = False    # Default — not passed as argument
        self.query_count = 0         # Internal tracking

    def connect(self):
        self.is_connected = True
        print(f"Connected to {self.database} at {self.host}:{self.port}")

# When you write this:
conn = DatabaseConnection("server.db.windows.net", 1433, "warehouse", "admin")

# Python actually does this behind the scenes:
# 1. Creates a new empty object
# 2. Calls __init__(new_object, "server.db...", 1433, "warehouse", "admin")
# 3. Returns the initialized object
# You NEVER call __init__ directly

Default Parameters in __init__

class PipelineConfig:
    def __init__(self, table_name, source="azure_sql", batch_size=10000,
                 retry_count=3, timeout=30, environment="production"):
        self.table_name = table_name
        self.source = source
        self.batch_size = batch_size
        self.retry_count = retry_count
        self.timeout = timeout
        self.environment = environment

# Only required parameter is table_name — rest have sensible defaults
config1 = PipelineConfig("customers")
config2 = PipelineConfig("orders", batch_size=50000, retry_count=5)
config3 = PipelineConfig("products", source="s3", environment="development")

print(config1.batch_size)    # 10000 (default)
print(config2.batch_size)    # 50000 (overridden)
print(config3.source)        # "s3" (overridden)

Validation in __init__

class PipelineConfig:
    VALID_SOURCES = {"azure_sql", "s3", "api", "sftp", "kafka"}
    VALID_ENVIRONMENTS = {"development", "staging", "production"}

    def __init__(self, table_name, source="azure_sql", batch_size=10000,
                 environment="production"):
        # Validate inputs at creation time — fail early
        if not table_name or not isinstance(table_name, str):
            raise ValueError(f"table_name must be a non-empty string, got: {table_name!r}")
        if source not in self.VALID_SOURCES:
            raise ValueError(f"Invalid source: {source}. Must be one of {self.VALID_SOURCES}")
        if batch_size < 1:
            raise ValueError(f"batch_size must be positive, got: {batch_size}")
        if environment not in self.VALID_ENVIRONMENTS:
            raise ValueError(f"Invalid environment: {environment}")

        self.table_name = table_name
        self.source = source
        self.batch_size = batch_size
        self.environment = environment

# Invalid config fails immediately — not halfway through a pipeline
try:
    bad_config = PipelineConfig("", source="ftp")
except ValueError as e:
    print(f"Config error: {e}")
    # Config error: table_name must be a non-empty string, got: ''

self — The Current Instance

What self Actually Is

# self is a reference to the SPECIFIC object calling the method
# When you call naveen.introduce(), Python passes naveen as self

class Employee:
    def __init__(self, name):
        self.name = name       # self.name = the name for THIS specific employee

    def greet(self):
        return f"Hello, I am {self.name}"  # self.name = THIS employee's name

naveen = Employee("Naveen")
alice = Employee("Alice")

naveen.greet()   # Python calls: Employee.greet(naveen) → self = naveen
alice.greet()    # Python calls: Employee.greet(alice)  → self = alice

# These two calls are identical:
naveen.greet()              # Shorthand (normal usage)
Employee.greet(naveen)      # What Python actually does

Why self Is Explicit in Python

In Java or C#, this is implicit — you do not write it in method signatures. In Python, self is explicit — you must include it as the first parameter of every instance method. This is a deliberate design choice: “explicit is better than implicit” (Zen of Python). It makes it clear that the method operates on a specific object, and it makes the data flow visible. The name self is a convention, not a keyword — you could call it anything, but always use self.

Attributes

Instance Attributes vs Class Attributes

class Pipeline:
    # CLASS ATTRIBUTE — shared by ALL instances, defined outside __init__
    version = "2.0"
    supported_formats = ["csv", "parquet", "json", "delta"]
    instance_count = 0

    def __init__(self, name, source):
        # INSTANCE ATTRIBUTES — unique to each object, defined in __init__
        self.name = name
        self.source = source
        self.status = "pending"
        self.row_count = 0
        Pipeline.instance_count += 1   # Modify the CLASS attribute

# Class attributes are accessed through the CLASS or any instance
print(Pipeline.version)           # "2.0"
print(Pipeline.supported_formats) # ["csv", "parquet", "json", "delta"]

p1 = Pipeline("customers", "azure_sql")
p2 = Pipeline("orders", "s3")

# Instance attributes are unique
print(p1.name)   # "customers"
print(p2.name)   # "orders"

# Class attributes are shared
print(p1.version)   # "2.0"
print(p2.version)   # "2.0"
print(Pipeline.instance_count)  # 2

When to Use Each

Use	Instance Attributes	Class Attributes
When	Each object needs its own value	All objects share the same value
Defined in	__init__ with self.	Class body (outside methods)
Examples	name, config, status, row_count	version, valid_sources, defaults, counters
Analogy	Each badge has a different name	All badges use the same template

The Mutable Class Attribute Trap

# ❌ DANGEROUS — mutable class attribute shared by ALL instances
class Pipeline:
    errors = []   # This list is SHARED — all instances append to the SAME list!

    def add_error(self, msg):
        self.errors.append(msg)

p1 = Pipeline()
p2 = Pipeline()

p1.add_error("File not found")
print(p2.errors)  # ["File not found"] — p2 sees p1's error! BUG!

# ✅ CORRECT — mutable attributes in __init__
class Pipeline:
    def __init__(self):
        self.errors = []   # Each instance gets its OWN list

    def add_error(self, msg):
        self.errors.append(msg)

p1 = Pipeline()
p2 = Pipeline()

p1.add_error("File not found")
print(p2.errors)  # [] — p2 has its own empty list. Correct!

# RULE: Never use mutable types (list, dict, set) as class attributes.
# Always create them in __init__ so each instance gets its own copy.

Methods

Instance Methods

# Instance methods — the most common type
# First parameter is self — they operate on a specific object

class TableLoader:
    def __init__(self, table_name, connection_string):
        self.table_name = table_name
        self.connection_string = connection_string
        self.row_count = 0

    def extract(self):
        """Instance method — reads from this loader's specific table."""
        print(f"Extracting from {self.table_name}")
        data = read_from_source(self.table_name, self.connection_string)
        self.row_count = len(data)
        return data

    def load(self, data, target):
        """Instance method — loads data and updates this loader's state."""
        write_to_target(data, target)
        print(f"Loaded {self.row_count:,} rows from {self.table_name} to {target}")

    def summary(self):
        """Instance method — reports on this specific loader."""
        return f"{self.table_name}: {self.row_count:,} rows loaded"

Class Methods (@classmethod)

# Class methods receive the CLASS as the first argument (cls), not an instance
# Use them as alternative constructors or for operations on the class itself

class PipelineConfig:
    def __init__(self, table_name, source, batch_size, environment):
        self.table_name = table_name
        self.source = source
        self.batch_size = batch_size
        self.environment = environment

    @classmethod
    def from_dict(cls, config_dict):
        """Create a PipelineConfig from a dictionary."""
        return cls(
            table_name=config_dict["table_name"],
            source=config_dict.get("source", "azure_sql"),
            batch_size=config_dict.get("batch_size", 10000),
            environment=config_dict.get("environment", "production")
        )

    @classmethod
    def from_json_file(cls, filepath):
        """Create a PipelineConfig from a JSON file."""
        import json
        with open(filepath, "r") as f:
            config_dict = json.load(f)
        return cls.from_dict(config_dict)

    @classmethod
    def dev_config(cls, table_name):
        """Quick dev config with sensible defaults."""
        return cls(table_name, source="local_csv", batch_size=100, environment="development")

# Usage — multiple ways to create the same type of object
config1 = PipelineConfig("customers", "azure_sql", 10000, "production")
config2 = PipelineConfig.from_dict({"table_name": "orders", "batch_size": 50000})
config3 = PipelineConfig.from_json_file("config.json")
config4 = PipelineConfig.dev_config("products")

Real-life analogy: Class methods are like different ways to order at a restaurant. You can order by telling the waiter exactly what you want (__init__), or point at a picture on the menu (from_dict), or say “I will have the lunch special” (dev_config). Each method creates the same type of meal — just different entry points.

Static Methods (@staticmethod)

# Static methods do NOT receive self or cls
# They are regular functions that live inside the class for organizational reasons
# They cannot access instance or class data

class DataValidator:
    def __init__(self, df):
        self.df = df

    @staticmethod
    def is_valid_email(email):
        """Utility function — does not need access to self or cls."""
        return isinstance(email, str) and "@" in email and "." in email

    @staticmethod
    def clean_phone_number(phone):
        """Remove non-digit characters from phone number."""
        return "".join(c for c in str(phone) if c.isdigit())

    def validate(self):
        """Instance method that uses the static methods."""
        bad_emails = self.df[~self.df["email"].apply(self.is_valid_email)]
        if len(bad_emails) > 0:
            print(f"Found {len(bad_emails)} invalid emails")

# Static methods can be called on the class OR an instance:
print(DataValidator.is_valid_email("naveen@example.com"))  # True
print(DataValidator.clean_phone_number("+1 (416) 555-0123"))  # "14165550123"

Choosing Between Method Types

Method Type	First Param	Can Access Instance?	Can Access Class?	Use For
Instance method	self	✅ Yes	✅ Yes	Most methods — operates on a specific object
@classmethod	cls	❌ No	✅ Yes	Alternative constructors, class-level operations
@staticmethod	None	❌ No	❌ No	Utility functions that belong to the class conceptually

Properties (@property)

Why Properties

Properties let you access a method as if it were an attribute. Instead of pipeline.get_status(), you write pipeline.status. This keeps the interface clean while allowing validation, computation, or side effects behind the scenes.

Getter, Setter, Deleter

class PipelineConfig:
    def __init__(self, table_name, batch_size=10000):
        self.table_name = table_name
        self._batch_size = batch_size   # "private" attribute (convention)

    @property
    def batch_size(self):
        """Getter — called when you ACCESS config.batch_size."""
        return self._batch_size

    @batch_size.setter
    def batch_size(self, value):
        """Setter — called when you SET config.batch_size = value."""
        if not isinstance(value, int) or value < 1:
            raise ValueError(f"batch_size must be a positive integer, got: {value}")
        if value > 1000000:
            raise ValueError(f"batch_size cannot exceed 1,000,000, got: {value}")
        self._batch_size = value

    @batch_size.deleter
    def batch_size(self):
        """Deleter — called when you DELETE config.batch_size."""
        print("Resetting batch_size to default")
        self._batch_size = 10000

# Usage — looks like a normal attribute, but has validation
config = PipelineConfig("customers")
print(config.batch_size)       # 10000 — calls the getter

config.batch_size = 50000      # Calls the setter — validates first
print(config.batch_size)       # 50000

config.batch_size = -1         # ValueError: batch_size must be a positive integer
del config.batch_size          # Resets to 10000

Computed Properties

class PipelineResult:
    def __init__(self, table_name, extracted, loaded, failed, duration):
        self.table_name = table_name
        self.extracted = extracted
        self.loaded = loaded
        self.failed = failed
        self.duration = duration

    @property
    def success_rate(self):
        """Computed on the fly — always up to date."""
        if self.extracted == 0:
            return 0.0
        return (self.loaded / self.extracted) * 100

    @property
    def rows_per_second(self):
        """Throughput metric."""
        if self.duration == 0:
            return 0
        return self.loaded / self.duration

    @property
    def has_failures(self):
        """Boolean computed property."""
        return self.failed > 0

result = PipelineResult("customers", extracted=50000, loaded=49950, failed=50, duration=12.5)
print(f"Success rate: {result.success_rate:.1f}%")      # 99.9%
print(f"Throughput: {result.rows_per_second:,.0f} rows/s")  # 3,996 rows/s
print(f"Has failures: {result.has_failures}")             # True

Dunder (Magic) Methods

Dunder (double underscore) methods let your objects work with Python’s built-in operators and functions — print(), len(), ==, [], with, and more. They make your objects feel like native Python types.

__str__ and __repr__

class TableResult:
    def __init__(self, table_name, row_count, status):
        self.table_name = table_name
        self.row_count = row_count
        self.status = status

    def __str__(self):
        """Human-readable string — used by print() and str()."""
        return f"{self.table_name}: {self.row_count:,} rows ({self.status})"

    def __repr__(self):
        """Developer-readable string — used in the REPL and debugging."""
        return f"TableResult('{self.table_name}', {self.row_count}, '{self.status}')"

result = TableResult("customers", 50000, "success")

print(result)        # customers: 50,000 rows (success)     — calls __str__
print(repr(result))  # TableResult('customers', 50000, 'success')  — calls __repr__
print([result])      # [TableResult('customers', 50000, 'success')] — lists use __repr__

# RULE: __repr__ should be unambiguous (ideally valid Python to recreate the object)
# RULE: __str__ should be readable (what you want the user to see)

__len__

class DataBatch:
    def __init__(self, records):
        self.records = records

    def __len__(self):
        """Called by len() — return the number of records."""
        return len(self.records)

batch = DataBatch([{"id": 1}, {"id": 2}, {"id": 3}])
print(len(batch))  # 3 — calls batch.__len__()

# Works with bool() too — empty = False, non-empty = True
if batch:
    print("Batch has data")

__eq__ and __lt__ (Comparison)

class TableResult:
    def __init__(self, table_name, row_count):
        self.table_name = table_name
        self.row_count = row_count

    def __eq__(self, other):
        """Called by == operator."""
        if not isinstance(other, TableResult):
            return NotImplemented
        return self.table_name == other.table_name and self.row_count == other.row_count

    def __lt__(self, other):
        """Called by < operator — enables sorting."""
        if not isinstance(other, TableResult):
            return NotImplemented
        return self.row_count < other.row_count

r1 = TableResult("customers", 50000)
r2 = TableResult("customers", 50000)
r3 = TableResult("orders", 100000)

print(r1 == r2)  # True
print(r1 == r3)  # False
print(r1 < r3)   # True (50000 < 100000)

# With __lt__ defined, sorted() works automatically:
results = [r3, r1, TableResult("products", 5000)]
for r in sorted(results):
    print(f"  {r.table_name}: {r.row_count:,}")
# products: 5,000
# customers: 50,000
# orders: 100,000

__getitem__ (Indexing)

class PipelineResults:
    def __init__(self):
        self._results = []

    def add(self, table_name, row_count, status):
        self._results.append({
            "table": table_name, "rows": row_count, "status": status
        })

    def __getitem__(self, index):
        """Called by [] operator — makes object indexable."""
        return self._results[index]

    def __len__(self):
        return len(self._results)

    def __iter__(self):
        """Makes object iterable — works with for loops."""
        return iter(self._results)

results = PipelineResults()
results.add("customers", 50000, "success")
results.add("orders", 100000, "success")
results.add("products", 5000, "failed")

print(results[0])           # {'table': 'customers', 'rows': 50000, 'status': 'success'}
print(results[-1])          # {'table': 'products', 'rows': 5000, 'status': 'failed'}
print(len(results))         # 3

for r in results:           # Works because __iter__ is defined
    print(f"  {r['table']}: {r['status']}")

__enter__ and __exit__ (Context Manager)

# __enter__ and __exit__ let your class work with the "with" statement
# Perfect for resources that need cleanup: connections, files, locks

class DatabaseConnection:
    def __init__(self, connection_string):
        self.connection_string = connection_string
        self.conn = None

    def __enter__(self):
        """Called when entering the with block — set up the resource."""
        print(f"Connecting to {self.connection_string}")
        self.conn = create_connection(self.connection_string)
        return self   # This becomes the "as" variable

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Called when exiting the with block — clean up, ALWAYS."""
        if self.conn:
            if exc_type:
                self.conn.rollback()   # Error occurred — undo changes
                print(f"Rolled back due to: {exc_val}")
            else:
                self.conn.commit()     # No error — save changes
            self.conn.close()
            print("Connection closed")
        return False   # False = do not suppress the exception

# Usage — connection is ALWAYS closed, even if an error occurs
with DatabaseConnection("server.db.windows.net") as db:
    db.conn.execute("INSERT INTO customers VALUES (1, 'Naveen')")
    db.conn.execute("INSERT INTO orders VALUES (101, 1, 99.99)")
# Connection automatically committed and closed here

# If an error occurs inside the with block:
# __exit__ is called → rollback → close → exception propagates

Encapsulation — Public, Protected, Private

Naming Conventions

class Pipeline:
    def __init__(self, name):
        self.name = name             # PUBLIC — accessible from anywhere
        self._connection = None      # PROTECTED — "please don't touch" (convention only)
        self.__secret_key = "abc"    # PRIVATE — name-mangled (harder to access)

    def connect(self):               # Public method
        self._setup_connection()     # Calls a protected method internally

    def _setup_connection(self):     # Protected — internal use
        self._connection = create_conn()

    def __reset_key(self):           # Private — name-mangled
        self.__secret_key = "new_key"

p = Pipeline("test")
print(p.name)              # ✅ Works — public
print(p._connection)       # ⚠️ Works — but you SHOULDN'T (convention)
# print(p.__secret_key)    # ❌ AttributeError — name-mangled!
print(p._Pipeline__secret_key)  # ✅ Works — but NEVER do this in real code

Name Mangling

Python does not have true private attributes like Java. The double underscore prefix (__name) triggers name mangling — Python renames it to _ClassName__name to avoid accidental access. The single underscore (_name) is purely a convention that says “this is internal, do not use it from outside.” Both can still be accessed — Python trusts you to respect the convention.

Real-life analogy: A public attribute is like your name badge — everyone can see it. A protected attribute (_) is like a staff-only door with a “Do Not Enter” sign — anyone can open it, but they should not. A private attribute (__) is like a door with the room number scratched off — you can still find it if you try, but Python is making it deliberately inconvenient.

Data Engineering Patterns

Pipeline Configuration Class

import json
from pathlib import Path

class PipelineConfig:
    """Immutable configuration for a pipeline run."""

    VALID_ENVIRONMENTS = {"development", "staging", "production"}

    def __init__(self, table_name, source_schema, target_schema,
                 batch_size=10000, environment="production"):
        self.table_name = table_name
        self.source_schema = source_schema
        self.target_schema = target_schema
        self._batch_size = batch_size
        self._environment = environment

    @property
    def batch_size(self):
        return self._batch_size

    @property
    def environment(self):
        return self._environment

    @property
    def full_source_name(self):
        return f"{self.source_schema}.{self.table_name}"

    @property
    def full_target_name(self):
        return f"{self.target_schema}.{self.table_name}"

    @classmethod
    def from_json(cls, filepath):
        with open(filepath, "r") as f:
            data = json.load(f)
        return cls(**data)

    @classmethod
    def from_dict(cls, d):
        return cls(**d)

    def __repr__(self):
        return (f"PipelineConfig('{self.table_name}', "
                f"'{self.source_schema}' → '{self.target_schema}', "
                f"batch={self.batch_size}, env='{self.environment}')")

# Usage
config = PipelineConfig("customers", "bronze", "silver", batch_size=50000)
print(config.full_source_name)  # "bronze.customers"
print(config.full_target_name)  # "silver.customers"
print(config)  # PipelineConfig('customers', 'bronze' → 'silver', batch=50000, env='production')

Table Loader Class

import logging
import time

class TableLoader:
    """Loads a single table through the extract-transform-load cycle."""

    def __init__(self, config):
        self.config = config
        self.logger = logging.getLogger(f"loader.{config.table_name}")
        self.row_count = 0
        self.duration = 0.0
        self.status = "pending"
        self.error = None

    def run(self):
        """Execute the full ETL cycle for this table."""
        start = time.time()
        self.logger.info(f"Starting: {self.config.full_source_name}")

        try:
            data = self._extract()
            transformed = self._transform(data)
            self._load(transformed)
            self.status = "success"
        except Exception as e:
            self.status = "failed"
            self.error = str(e)
            self.logger.error(f"Failed: {e}")
            raise
        finally:
            self.duration = time.time() - start
            self.logger.info(
                f"Finished: {self.status} | "
                f"{self.row_count:,} rows | {self.duration:.1f}s"
            )

    def _extract(self):
        self.logger.info("Extracting...")
        data = read_source(self.config.full_source_name)
        self.row_count = len(data)
        self.logger.info(f"Extracted {self.row_count:,} rows")
        return data

    def _transform(self, data):
        self.logger.info("Transforming...")
        return apply_transforms(data)

    def _load(self, data):
        self.logger.info(f"Loading to {self.config.full_target_name}...")
        write_target(data, self.config.full_target_name)

    def __repr__(self):
        return f"TableLoader({self.config.table_name}, {self.status}, {self.row_count:,} rows)"

Connection Manager (Context Manager)

import logging

class ConnectionManager:
    """Database connection with automatic cleanup via context manager."""

    def __init__(self, connection_string, autocommit=False):
        self.connection_string = connection_string
        self.autocommit = autocommit
        self.conn = None
        self.logger = logging.getLogger("db.connection")

    def __enter__(self):
        self.logger.info(f"Opening connection")
        self.conn = create_connection(self.connection_string)
        self.conn.autocommit = self.autocommit
        return self.conn

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.conn:
            if exc_type:
                self.conn.rollback()
                self.logger.error(f"Rolled back: {exc_val}")
            elif not self.autocommit:
                self.conn.commit()
                self.logger.info("Committed")
            self.conn.close()
            self.logger.info("Connection closed")
        return False

# Usage
with ConnectionManager(CONN_STRING) as conn:
    conn.execute("INSERT INTO audit_log VALUES (...)")
    conn.execute("UPDATE customers SET ...")
# Automatically committed and closed

Data Validator Class

class DataValidator:
    """Run validation checks on a dataset and collect results."""

    def __init__(self, table_name):
        self.table_name = table_name
        self.checks = []
        self.failures = []

    def check_not_null(self, df, column):
        """Validate that a column has no null values."""
        null_count = df[column].isnull().sum()
        passed = null_count == 0
        self.checks.append({
            "check": "not_null", "column": column,
            "passed": passed, "details": f"{null_count} nulls"
        })
        if not passed:
            self.failures.append(f"not_null({column}): {null_count} nulls")
        return self   # Enable chaining

    def check_unique(self, df, column):
        """Validate that a column has no duplicate values."""
        dupe_count = df[column].duplicated().sum()
        passed = dupe_count == 0
        self.checks.append({
            "check": "unique", "column": column,
            "passed": passed, "details": f"{dupe_count} duplicates"
        })
        if not passed:
            self.failures.append(f"unique({column}): {dupe_count} duplicates")
        return self

    def check_range(self, df, column, min_val, max_val):
        """Validate that values fall within a range."""
        out_of_range = ((df[column] < min_val) | (df[column] > max_val)).sum()
        passed = out_of_range == 0
        self.checks.append({
            "check": "range", "column": column,
            "passed": passed, "details": f"{out_of_range} out of [{min_val}, {max_val}]"
        })
        if not passed:
            self.failures.append(f"range({column}): {out_of_range} out of range")
        return self

    @property
    def is_valid(self):
        return len(self.failures) == 0

    @property
    def summary(self):
        total = len(self.checks)
        passed = sum(1 for c in self.checks if c["passed"])
        return f"{self.table_name}: {passed}/{total} checks passed"

    def raise_if_invalid(self):
        if not self.is_valid:
            raise ValueError(f"Validation failed for {self.table_name}: {self.failures}")

# Usage — method chaining makes it readable
validator = (DataValidator("customers")
    .check_not_null(df, "customer_id")
    .check_not_null(df, "email")
    .check_unique(df, "customer_id")
    .check_range(df, "age", 0, 150))

print(validator.summary)   # "customers: 3/4 checks passed"
if not validator.is_valid:
    print(f"Failures: {validator.failures}")

Pipeline Result Tracker

from datetime import datetime

class PipelineTracker:
    """Track results across multiple table loads."""

    def __init__(self, pipeline_name):
        self.pipeline_name = pipeline_name
        self.run_id = datetime.now().strftime("%Y%m%d_%H%M%S")
        self._results = []
        self._start_time = datetime.now()

    def record(self, table_name, row_count, status, duration, error=None):
        self._results.append({
            "table": table_name, "rows": row_count,
            "status": status, "duration": duration, "error": error
        })

    @property
    def total_rows(self):
        return sum(r["rows"] for r in self._results)

    @property
    def success_count(self):
        return sum(1 for r in self._results if r["status"] == "success")

    @property
    def failure_count(self):
        return sum(1 for r in self._results if r["status"] == "failed")

    @property
    def total_duration(self):
        return (datetime.now() - self._start_time).total_seconds()

    def __len__(self):
        return len(self._results)

    def __getitem__(self, index):
        return self._results[index]

    def __str__(self):
        return (
            f"Pipeline: {self.pipeline_name} (run: {self.run_id})
"
            f"  Tables: {self.success_count} success, {self.failure_count} failed
"
            f"  Rows: {self.total_rows:,}
"
            f"  Duration: {self.total_duration:.1f}s"
        )

# Usage
tracker = PipelineTracker("daily_load")
tracker.record("customers", 50000, "success", 3.2)
tracker.record("orders", 100000, "success", 8.5)
tracker.record("products", 0, "failed", 1.1, error="Connection timeout")

print(tracker)
print(f"Total tables: {len(tracker)}")
print(f"First result: {tracker[0]}")

Common Mistakes

1. Forgetting self in method definitions — def greet(): instead of def greet(self):. Python will pass the instance as the first argument, causing a “takes 0 positional arguments but 1 was given” error.
2. Forgetting self. when accessing attributes — writing name instead of self.name inside a method. Without self., Python looks for a local variable, not the instance attribute.
3. Mutable class attributes — using errors = [] as a class attribute instead of creating it in __init__. All instances share the same list, causing cross-contamination.
4. Not implementing __repr__ — without it, debugging shows <Pipeline object at 0x7f...> instead of useful information. Always define __repr__ for debuggability.
5. Returning None from __init__ — __init__ should never return a value. It initializes the object in place. If you need to return something, use a @classmethod instead.
6. Making every function a method — if a function does not use self or cls, it should be a @staticmethod or a standalone function, not an instance method. Not everything needs to be in a class.

Interview Questions

Q: What is the difference between a class and an instance? A: A class is a blueprint that defines attributes and methods. An instance is a specific object created from that blueprint. The class Employee defines what an employee looks like. naveen = Employee("Naveen") creates one specific employee. You can create many instances from one class, each with its own data.

Q: What is __init__ and what is self? A: __init__ is the constructor — it is called automatically when you create a new object. It sets up the object’s initial state by assigning values to attributes. self is a reference to the specific instance being created or operated on. Every instance method receives self as its first parameter, giving it access to that specific object’s data.

Q: What is the difference between instance methods, class methods, and static methods? A: Instance methods receive self and operate on a specific object’s data. Class methods receive cls and operate on the class itself — commonly used as alternative constructors. Static methods receive neither and are utility functions grouped under a class for organization. Use instance methods by default, class methods for alternative ways to create objects, and static methods for helper functions.

Q: What is the difference between instance attributes and class attributes? A: Instance attributes are defined in __init__ with self. and are unique to each object. Class attributes are defined in the class body and shared by all instances. Use instance attributes for data that varies per object (name, config) and class attributes for shared constants (version, valid values). Never use mutable types (list, dict) as class attributes.

Q: What are dunder methods and why are they useful? A: Dunder methods (double underscore, like __str__, __len__, __eq__) let your objects integrate with Python’s built-in operations. __str__ makes print() work. __len__ makes len() work. __eq__ makes == work. __enter__ and __exit__ make the with statement work. They let custom objects behave like native Python types.

Q: What is a property and when should you use one? A: A property (@property) lets you access a method as if it were an attribute. Use it for computed values (success_rate), validated setters (reject invalid batch_size), or read-only attributes. It keeps the interface clean — result.success_rate instead of result.get_success_rate() — while allowing logic behind the scenes.

Wrapping Up

OOP in Python gives you a way to bundle data and behavior together — and that matters when your pipeline has dozens of tables, each with their own config, transformation logic, and validation rules. Classes give you reusable blueprints. __init__ sets up each object with its own state. Methods define what each object can do. Properties compute values on the fly. Dunder methods make your objects work with Python’s built-in operators.

The data engineering patterns here — PipelineConfig, TableLoader, ConnectionManager, DataValidator, PipelineTracker — are patterns you will use in real production pipelines. They keep configuration separate from logic, ensure resources are cleaned up, and make pipeline results trackable and reportable.

Previous in this series: Logging — Levels, Formatters, Handlers, and Production Patterns

Next in this series: OOP Advanced — Inheritance, Polymorphism, Abstract Classes

---

Naveen Vuppula is a Senior Data Engineering Consultant and app developer based in Ontario, Canada. He writes about Python, SQL, AWS, Azure, and everything data engineering at DriveDataScience.com.