I'm trying to understand the DDD architecture pattern. I wrote a simple project in which I tried to use DDD arch. Here are my doubts after implementing it:
- Does it make sense to use an ORM for a DDD project? Maybe I should drop the use of Base class from declarative_base?
- Where should I define validators - in domain objects or in repository objects?
- Should
SimpleService(session)take a session in the initializer or specific repositories that I will create in the main function? Now I create it here:
class SimpleService: """Creates simple service.""" def __init__(self, session): self.session = session self.post_repository = PostRepository(session) self.comment_repository = CommentRepository(session)
- What should I change or improve in whole project? Please give me a quick review.
Short summary of project
Project structure:
.
│ main.py
│ README.rst
│ requirements.txt
│
├───blog
│ │ __init__.py
│ │
│ ├───domain
│ │ comment.py
│ │ post.py
│ │ __init__.py
│ │
│ ├───repository
│ │ │ __init__.py
│ │ │
│ │ ├───crud
│ │ │ base.py
│ │ │ comment.py
│ │ │ post.py
│ │ │ __init__.py
│ │ │
│ │ └───models
│ │ base.py
│ │ comment.py
│ │ post.py
│ │ __init__.py
│ │
│ └───service
│ simple.py
│ __init__.py
Business model in domain/post.py:
from dataclasses import dataclass
# from .comment import Comment
@dataclass
class Post:
id: str
title: str
content: str
# Should I implement it? If yes how to implement it here and in crud?
# comments: list[Comment]
#
# def __post_init__(self):
# if self.comments is None:
# self.comments = []
Orm model in repository/models/post.py:
from sqlalchemy import Column, String
from sqlalchemy.orm import relationship
from .base import Base
class Post(Base):
__tablename__ = 'posts'
id = Column(String, primary_key=True)
title = Column(String)
content = Column(String)
executions = relationship('Comment', backref='post')
def __repr__(self):
return f'Post(id={self.id}, title={self.title}, content={self.content})'
CRUD class to manage orm model repository/crud/post.py:
from blog.repository.crud.base import BaseRepository
from blog.repository.models.post import Post as PostModel
from blog.domain.post import Post as PostEntity
class PostRepository(BaseRepository):
"""Creates object to manage post in data in database."""
def __init__(self, session):
self.session = session
def create(self, post):
"""Creates post."""
post_model = PostModel(id=post.id, title=post.title, content=post.content)
self.session.add(post_model)
self.session.commit()
def get_by_id(self, id_):
"""Gets post by id."""
post_model = self.session.query(PostModel).get(id_)
if post_model is not None:
return PostEntity(id=post_model.id, title=post_model.title, content=post_model.content)
Main business logic in service/simple.py:
import time
import uuid
from blog.repository.crud.comment import CommentRepository
from blog.repository.crud.post import PostRepository
from blog.domain.post import Post
from blog.domain.comment import Comment
class SimpleService:
"""Creates simple service."""
def __init__(self, session):
self.session = session
self.post_repository = PostRepository(session)
self.comment_repository = CommentRepository(session)
def run(self):
"""Performs fake operation (main app logic)."""
print('Fake action in simple service...')
print('Creating posts...')
time.sleep(1)
post_id = str(uuid.uuid4())
post = Post(id=post_id,
title='Post 1',
content='Lorem Ipsum is simply dummy text of the printing and typesetting industry.')
self.post_repository.create(post)
print('Creating comment...')
time.sleep(1)
comment_id = str(uuid.uuid4())
comment_1 = Comment(id=comment_id,
content='Lorem Ipsum has been the industry\'s standard dummy text',
post=post)
self.comment_repository.create(comment_1)
print('Getting post from db...')
time.sleep(1)
print(self.post_repository.get_by_id(post_id))
print('Getting comment from db...')
time.sleep(1)
print(self.comment_repository.get_by_id(comment_id))
Entrypoint of app in main.py:
import os
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from blog.repository.models.base import Base
from blog.service.simple import SimpleService
def main():
"""Entrypoint of app."""
database_uri = os.getenv('DATABASE_URI') # set DATABASE_URI=sqlite:///data.db
if database_uri is None:
database_uri = 'sqlite:///:memory:'
engine = create_engine(database_uri)
Base.metadata.create_all(engine)
with Session(engine) as session:
simple_service = SimpleService(session)
simple_service.run()
if __name__ == '__main__':
main()
database_uri = os.getenv('DATABASE_URI', 'sqlite:///:memory:')\$\endgroup\$