SystemSh0cker/private-gpt
1
0
Fork 0
mirror of https://github.com/zylon-ai/private-gpt.git synced 2025-12-22 20:12:55 +01:00
Code Issues Projects Releases Packages Activity

update at documents router

Browse source
This commit is contained in:
Saurab-Shrestha 2024-03-19 16:24:52 +05:45
parent 2008837110
commit c7c05de8d1
12 changed files with 264 additions and 24 deletions
Download patch file Download diff file Expand all files Collapse all files

46
private_gpt/users/crud/document_crud.py
View file

@ -1,3 +1,4 @@
from sqlalchemy.sql.expression import desc, asc
from sqlalchemy import or_, and_
from sqlalchemy.orm import Session
from private_gpt.users.schemas.documents import DocumentCreate, DocumentUpdate
@ -16,11 +17,11 @@ class CRUDDocuments(CRUDBase[Document, DocumentCreate, DocumentUpdate]):
return db.query(self.model).filter(Document.filename == file_name).first()
def get_multi_documents(
self, db: Session, *,department_id: int, skip: int = 0, limit: int = 100
self, db: Session, *, skip: int = 0, limit: int = 100
) -> List[Document]:
return (
db.query(self.model)
.filter(Document.department_id == department_id)
.order_by(desc(getattr(Document, 'uploaded_at')))
.offset(skip)
.limit(limit)
.all()
@ -36,13 +37,24 @@ class CRUDDocuments(CRUDBase[Document, DocumentCreate, DocumentUpdate]):
.filter(document_department_association.c.department_id == department_id)
.offset(skip)
.limit(limit)
.all().order_by(desc(getattr(Document, 'uploaded_at')))
)
def get_files_to_verify(
self, db: Session, *, skip: int = 0, limit: int = 100
) -> List[Document]:
return (
db.query(self.model)
.filter(Document.status == 'PENDING')
.offset(skip)
.limit(limit)
.all()
)
def get_enabled_documents_by_departments(
self, db: Session, *, department_id: int, skip: int = 0, limit: int = 100
) -> List[Document]:
all_department_id = 4 # department ID for "ALL" is 4
all_department_id = 1 # department ID for "ALL" is 1
return (
db.query(self.model)
@ -65,5 +77,33 @@ class CRUDDocuments(CRUDBase[Document, DocumentCreate, DocumentUpdate]):
.all()
)
def filter_query(
self, db: Session, *,
filename: Optional[str] = None,
uploaded_by: Optional[str] = None,
action_type: Optional[str] = None,
status: Optional[str] = None,
order_by: Optional[str] = None,
skip: int = 0,
limit: int = 100
) -> List[Document]:
query = db.query(Document)
if filename:
query = query.filter(
Document.filename.ilike(f"%{filename}%"))
if uploaded_by:
query = query.filter(
Document.uploaded_by == uploaded_by)
if action_type:
query = query.filter(
Document.action_type == action_type)
if status:
query = query.filter(Document.status == status)
if order_by == "desc":
query = query.order_by(desc(getattr(Document, 'uploaded_at')))
else:
query = query.order_by(asc(getattr(Document, 'uploaded_at')))
return query.offset(skip).limit(limit).all()
documents = CRUDDocuments(Document)