Add Items (crud, models, endpoints), utils, refactor (#14)

* Update CRUD utils to use types better.
* Simplify Pydantic model names, from `UserInCreate` to `UserCreate`, etc.
* Upgrade packages.
* Add new generic "Items" models, crud utils, endpoints, and tests. To facilitate re-using them to create new functionality. As they are simple and generic (not like Users), it's easier to copy-paste and adapt them to each use case.
* Update endpoints/*path operations* to simplify code and use new utilities, prefix and tags in `include_router`.
* Update testing utils.
* Update linting rules, relax vulture to reduce false positives.
* Update migrations to include new Items.
* Update project README.md with tips about how to start with backend.
This commit is contained in:
Sebastián Ramírez
2019-04-19 09:45:23 +04:00
committed by GitHub
parent 1fe4908b0a
commit ecd634e497
32 changed files with 426 additions and 1091 deletions

View File

@@ -1 +1 @@
from . import user
from . import item, user

View File

@@ -0,0 +1,54 @@
from typing import List, Optional
from fastapi.encoders import jsonable_encoder
from sqlalchemy.orm import Session
from app.db_models.item import Item
from app.models.item import ItemCreate, ItemUpdate
def get(db_session: Session, *, id: int) -> Optional[Item]:
return db_session.query(Item).filter(Item.id == id).first()
def get_multi(db_session: Session, *, skip=0, limit=100) -> List[Optional[Item]]:
return db_session.query(Item).offset(skip).limit(limit).all()
def get_multi_by_owner(
db_session: Session, *, owner_id: int, skip=0, limit=100
) -> List[Optional[Item]]:
return (
db_session.query(Item)
.filter(Item.owner_id == owner_id)
.offset(skip)
.limit(limit)
.all()
)
def create(db_session: Session, *, item_in: ItemCreate, owner_id: int) -> Item:
item = Item(title=item_in.title, description=item_in.description, owner_id=owner_id)
db_session.add(item)
db_session.commit()
db_session.refresh(item)
return item
def update(db_session: Session, *, item: Item, item_in: ItemUpdate) -> Item:
item_data = jsonable_encoder(item)
update_data = item_in.dict(skip_defaults=True)
for field in item_data:
if field in update_data:
setattr(item, field, update_data[field])
db_session.add(item)
db_session.commit()
db_session.refresh(item)
return item
def remove(db_session: Session, *, id: int):
item = db_session.query(Item).filter(Item.id == id).first()
db_session.delete(item)
db_session.commit()
return item

View File

@@ -1,21 +1,22 @@
from typing import List, Optional
from fastapi.encoders import jsonable_encoder
from sqlalchemy.orm import Session
from app.core.security import get_password_hash, verify_password
from app.db_models.user import User
from app.models.user import UserInCreate, UserInUpdate
from app.models.user import UserCreate, UserUpdate
def get(db_session, *, user_id: int) -> Optional[User]:
def get(db_session: Session, *, user_id: int) -> Optional[User]:
return db_session.query(User).filter(User.id == user_id).first()
def get_by_email(db_session, *, email: str) -> Optional[User]:
def get_by_email(db_session: Session, *, email: str) -> Optional[User]:
return db_session.query(User).filter(User.email == email).first()
def authenticate(db_session, *, email: str, password: str) -> Optional[User]:
def authenticate(db_session: Session, *, email: str, password: str) -> Optional[User]:
user = get_by_email(db_session, email=email)
if not user:
return None
@@ -32,11 +33,11 @@ def is_superuser(user) -> bool:
return user.is_superuser
def get_multi(db_session, *, skip=0, limit=100) -> List[Optional[User]]:
def get_multi(db_session: Session, *, skip=0, limit=100) -> List[Optional[User]]:
return db_session.query(User).offset(skip).limit(limit).all()
def create(db_session, *, user_in: UserInCreate) -> User:
def create(db_session: Session, *, user_in: UserCreate) -> User:
user = User(
email=user_in.email,
hashed_password=get_password_hash(user_in.password),
@@ -49,13 +50,12 @@ def create(db_session, *, user_in: UserInCreate) -> User:
return user
def update(db_session, *, user: User, user_in: UserInUpdate) -> User:
def update(db_session: Session, *, user: User, user_in: UserUpdate) -> User:
user_data = jsonable_encoder(user)
update_data = user_in.dict(skip_defaults=True)
for field in user_data:
if field in user_in.fields:
value_in = getattr(user_in, field)
if value_in is not None:
setattr(user, field, value_in)
if field in update_data:
setattr(user, field, update_data[field])
if user_in.password:
passwordhash = get_password_hash(user_in.password)
user.hashed_password = passwordhash