surtur
fc0051b243
* aside from slight refactors there have been route changes (/api/v1/service/ --> /api/v1/service) * added helper little function for grabbing user and service ids * refactored a couple of return statements * added separate models for updates * adjusted error handling behaviour: more general 422 is now returned (previously 404 was used) --> hopefully leaking less information * updates are divided into multiple queries **as of now**, mainly due to my sqlalchemy amateurism: can't figure out how to construct a proper multi-column-update query programatically
93 lines
3.7 KiB
Python
93 lines
3.7 KiB
Python
import bcrypt, time
|
|
from sqlalchemy.orm import Session
|
|
from . import models, schemas
|
|
|
|
|
|
def get_user(db: Session, user_id: int):
|
|
return db.query(models.User).filter(models.User.id == user_id).first()
|
|
|
|
|
|
def del_user(db: Session, user_id: int):
|
|
db.query(models.User).filter(models.User.id == user_id).delete()
|
|
return db.commit()
|
|
|
|
|
|
def get_user_by_username(db: Session, name: str):
|
|
return db.query(models.User).filter(models.User.name == name).first()
|
|
|
|
|
|
def get_users(db: Session, skip: int = 0, limit: int = 100):
|
|
return db.query(models.User).offset(skip).limit(limit).all()
|
|
|
|
|
|
def create_user(db: Session, user: schemas.UserCreate):
|
|
user.password = bcrypt.hashpw(user.password.encode(), bcrypt.gensalt())
|
|
db_user = models.User(**user.dict())
|
|
db_user.created_unix = time.time()
|
|
db.add(db_user)
|
|
db.commit()
|
|
db.refresh(db_user)
|
|
return db_user
|
|
|
|
|
|
def update_user(db: Session, user_id: int, user: schemas.UserUpdate):
|
|
updated_user = models.User(**user.dict())
|
|
if (updated_user.password is not None):
|
|
updated_user.password = bcrypt.hashpw(user.password.encode(), bcrypt.gensalt())
|
|
db.query(models.User).filter(models.User.id==user_id).update({models.User.password: updated_user.password})
|
|
if (updated_user.name is not None):
|
|
db.query(models.User).filter(models.User.id==user_id).update({models.User.name: updated_user.name})
|
|
if (updated_user.full_name is not None):
|
|
db.query(models.User).filter(models.User.id==user_id).update({models.User.full_name: updated_user.full_name})
|
|
if (updated_user.is_active is not None):
|
|
db.query(models.User).filter(models.User.id==user_id).update({models.User.is_active: updated_user.is_active})
|
|
if (updated_user is not user):
|
|
db.query(models.User).filter(models.User.id==user_id).update({models.User.updated_unix: time.time()})
|
|
return db.commit()
|
|
|
|
|
|
|
|
|
|
def get_service(db: Session, service_id: int):
|
|
return db.query(models.Service).filter(models.Service.id == service_id).first()
|
|
|
|
|
|
def get_services(db: Session, skip: int = 0, limit: int = 100):
|
|
return db.query(models.Service).offset(skip).limit(limit).all()
|
|
|
|
|
|
def create_service(db: Session, service: schemas.ServiceCreate):
|
|
db_item = models.Service(**service.dict())
|
|
db_item.created_unix = time.time()
|
|
db.add(db_item)
|
|
db.commit()
|
|
db.refresh(db_item)
|
|
return db_item
|
|
|
|
|
|
def del_service(db: Session, service_id: int):
|
|
db.query(models.Service).filter(models.Service.id == service_id).delete()
|
|
return db.commit()
|
|
|
|
|
|
def update_service(db: Session, service_id: int, s: schemas.ServiceUpdate):
|
|
updated_s = models.Service(**s.dict())
|
|
if (updated_s.name is not None):
|
|
db.query(models.Service).filter(models.Service.id==service_id).update({models.Service.name: updated_s.name})
|
|
if (updated_s.owner_id is not None):
|
|
db.query(models.Service).filter(models.Service.id==service_id).update({models.Service.owner_id: updated_s.owner_id})
|
|
if (updated_s.is_private is not None):
|
|
db.query(models.Service).filter(models.Service.id==service_id).update({models.Service.is_private: updated_s.is_private})
|
|
if (updated_s.description is not None):
|
|
db.query(models.Service).filter(models.Service.id==service_id).update({models.Service.description: updated_s.description})
|
|
if (updated_s.service_type is not None):
|
|
db.query(models.Service).filter(models.Service.id==service_id).update({models.Service.service_type: updated_s.service_type})
|
|
if (updated_s.url is not None):
|
|
db.query(models.Service).filter(models.Service.id==service_id).update({models.Service.url: updated_s.url})
|
|
if (updated_s.is_active is not None):
|
|
db.query(models.Service).filter(models.Service.id==service_id).update({models.Service.is_active: updated_s.is_active})
|
|
if (updated_s is not s):
|
|
db.query(models.Service).filter(models.Service.id==service_id).update({models.Service.updated_unix: time.time()})
|
|
return db.commit()
|
|
|