first

src/personal_info/services/__init__.py

@@ -0,0 +1,11 @@
+from fastapi import Depends
+from sqlalchemy.orm import Session
+
+from personal_info.db.session import get_session
+
+from .people import PersonService
+
+def get_persons_service(db_session: Session = Depends(get_session)) -> PersonService:
+  return PersonService(db_session)
+
+__all__ = ('get_persons_service')

src/personal_info/services/base.py

@@ -0,0 +1,52 @@
+from typing import Any, Generic, List, Optional, Type, TypeVar
+
+import sqlalchemy
+from pydantic import BaseModel
+from sqlalchemy.orm import Session
+from starlette.exceptions import HTTPException
+
+from personal_info.db.models import Base
+
+ModelType = TypeVar('ModelType', bound=Base)
+CreateSchemaType = TypeVar('CreateSchemaType', bound=BaseModel)
+UpdateSchemaType = TypeVar('UpdateSchemaType', bound=BaseModel)
+
+class BaseService(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
+  def __init__(self, model: Type[ModelType], db_session: Session):
+    self.model = model
+    self.db_session = db_session
+
+  def get(self, id: Any) -> Optional[ModelType]:
+    obj: Optional[ModelType] = self.db_session.get(self.model, id)
+    if obj is None:
+      raise HTTPException(status_code=404, detail='NotFound')
+    return obj
+
+  def list(self) -> List[ModelType]:
+    objs: List[ModelType] = self.db_session.query(self.model).all()
+    return objs
+
+  def create(self, obj: CreateSchemaType) -> ModelType:
+    db_obj: ModelType = self.model(**obj.dict())
+    self.db_session.add(db_obj)
+    try:
+      self.db_session.commit()
+    except sqlalchemy.exc.IntegrityError as e:
+      self.db_session.rollback()
+      if 'duplicate key' in str(e):
+        raise HTTPException(status_code=409, detail='Conflict Error')
+      else:
+        raise e
+    return db_obj
+
+  def update(self, id: Any, obj: UpdateSchemaType) -> Optional[ModelType]:
+    db_obj = self.get(id)
+    for column, value in obj.dict(exclude_unset=True).items():
+      setattr(db_obj, column, value)
+    self.db_session.commit()
+    return db_obj
+
+  def delete(self, id: Any) -> None:
+    db_obj = self.db_session.get(self.model, id)
+    self.db_session.delete(db_obj)
+    self.db_session.commit()

src/personal_info/services/people.py

@@ -0,0 +1,19 @@
+from typing import Any
+
+from sqlalchemy.orm import Session
+
+from personal_info.db.models import Person
+from personal_info.db.schemas import PersonCreate
+
+from personal_info.services.base import BaseService
+
+class PersonService(BaseService[Person, PersonCreate, Any]):
+  def __init__(self, db_session: Session):
+    super(PersonService, self).__init__(Person, db_session)
+
+  def create(self, obj: PersonCreate) -> Person:
+    person: Person = Person()
+    self.db_session.add(person)
+    self.db_session.flush()
+    self.db_session.commit()
+    return person