If you’ve got multiple data sources that you’re pulling from: an s3 bucket, a postgres database, and an API, the code can be relatively chaotic if you’re not managing it correctly.
You might have a function call to ingest from an s3 bucket, another function call to ingest from the API, and then likely a bunch of if/else statements:
def ingest_data(source: str):
if source == 'api':
data = ingest_from_api()
elif source == 's3':
data = ingest_from_s3()
else:
data = ingest_from_postgres()
return dataWhile the code looks valid, there are multiple issues:
Unknown return data types. We don’t know the data type/structure for the return data type - is it bytes? JSON? String?
Adding a data source requires updating. While it seems harmless, what if we want to add a mock data source? What about a mock S3 bucket? This if/else pattern requires us to touch this specific function, which could introduce breaking changes.
The else statement is catch-all. What if we pass in
postgress? What aboutcsv? This function requires us to know what data sources are available.
One of my favorite software patterns to manage this issue is called the Repository Pattern. This pattern puts one class per data source behind a single shared interface so that the rest of your code calls the same method no matter which data source you’re using.
The Repository Pattern
The pattern is simple: define a single class with a .get method that every data source implements. We’ll be inheriting from this class:
from abc import ABC, abstractmethod
class DataSource(ABC):
@abstractmethod
def get(*args, **kwargs) -> bytes:
...» The *args and **kwargs is important here. Don’t know what these do? Refer to this guide.
From here, we’ll define each data source as its own class implementing its own .get method:
# S3 data source
class S3DataSource(DataSource):
def __init__(self, bucket: str):
self.client = boto3.client("s3")
self.bucket = bucket
def get(self, doc_id: str) -> bytes:
obj = self.client.get_object(
Bucket=self.bucket, Key=doc_id
)
return obj["Body"].read()
# Postgres data source
class PostgresDataSource(DataSource):
def __init__(self, session):
self.session = session
def get(self, doc_id: str) -> bytes:
row = self.session.query(
Document
).filter_by(id=doc_id).first()
return row.content.encode()
# API data source
class APIDataSource(DataSource):
def __init__(self, base_url: str):
self.base_url = base_url
def get(self, doc_id: str) -> bytes:
resp = requests.get(f"{self.base_url}/{doc_id}")
return resp.json()["content"].encode()Now that you have each data source as their own objects with their own .get methods, you’ll be able to tie it all together:
SOURCES = {
's3' : S3DataSource(bucket = "docs-bucket"),
'postgres' : PostgresDataSource(session = get_db_session()),
'api' : APIDataSource(url = "https://my-api.com/api")
}
# This is the function for others to call
def get_data(source: str, *args, **kwargs) -> bytes:
repo = SOURCES.get(source)
if not repo:
print(f"{source} is not a valid source")
return 0
return repo.get(*args, **kwargs)» 2 weeks ago, I had sent out the registry pattern. We do implement this here with the SOURCES variable.
Notice that no matter what data source is used, it first checks to make sure it’s a valid data source within SOURCES, then it calls the appropriate get method. If we wanted to stand up a new source, we’d follow the same pattern:
Identify the source (i.e. a mock source)
Write the
getmethod logic to fetch the data and return the data asbytes.Add it to
SOURCESdictionary.
So while yes, there’s a significant more amount of code added and it looks more complicated, it pays off because you can easily add/remove data sources with a single line of code while never touching client-side code.
Want your AI agent to already know patterns like this one?
Happy coding!
📧 Join the Python Snacks Newsletter! 🐍
Want even more Python-related content that’s useful? Here’s 3 reasons why you should subscribe the Python Snacks newsletter:
Get Ahead in Python with bite-sized Python tips and tricks delivered straight to your inbox, like the one above.
Exclusive Subscriber Perks: Receive a curated selection of up to 6 high-impact Python resources, tips, and exclusive insights with each email.
Get Smarter with Python in under 5 minutes. Your next Python breakthrough could just an email away.
You can unsubscribe at any time.
Interested in starting a newsletter or a blog?
Do you have a wealth of knowledge and insights to share with the world? Starting your own newsletter or blog is an excellent way to establish yourself as an authority in your field, connect with a like-minded community, and open up new opportunities.
If TikTok, Twitter, Facebook, or other social media platforms were to get banned, you’d lose all your followers. This is why you should start a newsletter: you own your audience.
This article may contain affiliate links. Affiliate links come at no cost to you and support the costs of this blog. Should you purchase a product/service from an affiliate link, it will come at no additional cost to you.
