148 lines
6.3 KiB
Python
148 lines
6.3 KiB
Python
from datetime import datetime
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import select,and_
|
|
import typing as t
|
|
|
|
from app.db.cruddb.crudbase import crudbase
|
|
from fastapi_pagination.ext.sqlalchemy import paginate
|
|
from app.core.common import ApiReturnPage
|
|
|
|
from app.db import models, schemas
|
|
from app.core.security import chacha20Decrypt, get_password_hash
|
|
|
|
class dbuserdomain(crudbase):
|
|
def __init__(self):
|
|
super().__init__(model=models.UserDomain)
|
|
|
|
def get_userdomain(self,db: Session,userid:int,domainid:int):
|
|
return db.execute(super().get_by_conditions({"userid":userid,"domainid":domainid})).scalars().first()
|
|
|
|
def get_userdomain_by_domainid(self,db: Session,ownerid:int,domainid:int):
|
|
return super().get_by_conditions({"domainid":domainid})
|
|
|
|
|
|
def get_default_domains(self,db: Session,domainid:int):
|
|
return db.execute(super().get_by_conditions({"domainid":domainid,"is_default":True})).scalars().all()
|
|
|
|
def get_user_default_domain(self,db: Session,userid:int):
|
|
return db.execute(super().get_by_conditions({"userid":userid,"is_default":True})).scalars().first()
|
|
|
|
|
|
dbuserdomain = dbuserdomain()
|
|
|
|
class dbdomain(crudbase):
|
|
def __init__(self):
|
|
super().__init__(model=models.Domain)
|
|
|
|
def get_domains(self,db: Session)-> ApiReturnPage[models.Base]:
|
|
return paginate(db,super().get_all())
|
|
|
|
def get_domains_by_owner(self,db: Session,ownerid:int)-> ApiReturnPage[models.Base]:
|
|
return paginate(db,super().get_by_conditions({"ownerid":ownerid}))
|
|
|
|
def create_domain(self,db: Session, domain: schemas.DomainIn,userid:int):
|
|
#db_domain = super().get_by_conditions(db,{"url":domain.url,"kintoneuser":domain.kintoneuser,"onwerid":userid}).first()
|
|
#if not db_domain:
|
|
domain.encrypt_kintonepwd()
|
|
domain.id = None
|
|
domain.createuserid = userid
|
|
domain.updateuserid = userid
|
|
domain.ownerid = userid
|
|
return super().create(db,domain)
|
|
#return db_domain
|
|
|
|
def delete_domain(self,db: Session,id: int):
|
|
return super().delete(db,id)
|
|
|
|
def edit_domain(self,db: Session, domain: schemas.DomainIn,userid:int) -> schemas.DomainOut:
|
|
db_domain = super().get(db,domain.id)
|
|
if db_domain:
|
|
db_domain.tenantid = domain.tenantid
|
|
db_domain.name=domain.name
|
|
db_domain.url=domain.url
|
|
if db_domain.is_active == True and domain.is_active == False:
|
|
db_userdomains = dbuserdomain.get_default_domains(db,domain.id)
|
|
for userdomain in db_userdomains:
|
|
userdomain.is_default = False
|
|
db.add(userdomain)
|
|
db_domain.is_active=domain.is_active
|
|
db_domain.kintoneuser=domain.kintoneuser
|
|
if domain.kintonepwd != "" and domain.kintonepwd != None:
|
|
domain.encrypt_kintonepwd()
|
|
db_domain.kintonepwd = domain.kintonepwd
|
|
db_domain.updateuserid = userid
|
|
db.add(db_domain)
|
|
db.commit()
|
|
db.refresh(db_domain)
|
|
return db_domain
|
|
return None
|
|
|
|
def add_userdomain(self,db: Session,ownerid:int,userid:int,domainid:int) -> schemas.DomainOut:
|
|
db_domain = super().get(db,domainid)
|
|
if db_domain and db_domain.is_active:
|
|
db_userdomain = dbuserdomain.get_userdomain(db,userid,domainid)
|
|
if not db_userdomain:
|
|
user_domain = models.UserDomain(userid = userid, domainid = domainid ,createuserid = ownerid,updateuserid = ownerid)
|
|
db.add(user_domain)
|
|
db.commit()
|
|
return db_domain
|
|
return None
|
|
|
|
def add_userdomain_by_owner(self,db: Session,ownerid:int, userid:int,domainid:int) -> schemas.DomainOut:
|
|
db_domain = super().get_by_conditions(db,{"id":domainid,"ownerid":ownerid,"is_active":True}).first()
|
|
if db_domain:
|
|
db_userdomain = dbuserdomain.get_userdomain(db,userid,domainid)
|
|
if not db_userdomain:
|
|
user_domain = models.UserDomain(userid = userid, domainid = domainid ,createuserid =ownerid,updateuserid = ownerid)
|
|
db.add(user_domain)
|
|
db.commit()
|
|
return db_domain
|
|
return None
|
|
|
|
def delete_userdomain(self,db: Session, userid: int,domainid: int) -> schemas.DomainOut:
|
|
db_userdomain = dbuserdomain.get_userdomain(db,userid,domainid)
|
|
if db_userdomain:
|
|
domain = db_userdomain.domain
|
|
db.delete(db_userdomain)
|
|
db.commit()
|
|
return domain
|
|
return None
|
|
|
|
def get_default_domain(self,db: Session, userid: int) -> schemas.DomainOut:
|
|
userdomain = dbuserdomain.get_user_default_domain(db,userid)
|
|
if userdomain:
|
|
return userdomain.domain
|
|
else:
|
|
return None
|
|
|
|
def set_default_domain(self,db: Session, userid: int,domainid: int):
|
|
db_domain =db.execute(super().get_by_conditions({"id":domainid,"is_active":True})).scalars().first()
|
|
if db_domain:
|
|
db_default_domain = dbuserdomain.get_user_default_domain(db,userid)
|
|
db_userdomain =dbuserdomain.get_userdomain(db,userid,domainid)
|
|
if db_default_domain:
|
|
if db_default_domain.domainid != domainid:
|
|
db_default_domain.is_default = False
|
|
db_default_domain.updateuserid = userid
|
|
db.add(db_default_domain)
|
|
else:
|
|
return db_domain
|
|
if db_userdomain:
|
|
db_userdomain.is_default = True
|
|
db_userdomain.updateuserid = userid
|
|
db.add(db_userdomain)
|
|
else:
|
|
db_userdomain = dbuserdomain.create(db,schemas.UserDomainIn(domainid=domainid,userid=userid,is_default = True))
|
|
db.add(db_userdomain)
|
|
db.commit()
|
|
return db_domain
|
|
else:
|
|
return None
|
|
|
|
|
|
def get_shareddomain_users(self,db: Session,ownerid:int,domainid: int) -> ApiReturnPage[models.Base]:
|
|
users = select(models.User).join(models.UserDomain,models.UserDomain.userid == models.User.id).filter(models.UserDomain.domainid ==domainid)
|
|
return paginate(db,users)
|
|
|
|
domainService = dbdomain() |