bolade/Anton_wireframe
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Implement pagination for companies, investors, and projects endpoints

Browse Source
This commit is contained in:
bolade
2025-10-08 10:25:52 +01:00
parent 26a1197db0
commit faf92a3b47
4 changed files with 190 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/routers/companies.py
+53 -14
View File
@@ -1,10 +1,10 @@
from typing import List, Optional
from typing import Optional
from db.db import get_db
from db.models import CompanyTable, InvestorTable
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel
from schemas.router_schemas import CompanyData
from schemas.router_schemas import CompanyData, PaginatedResponse
from sqlalchemy.orm import Session, selectinload
router = APIRouter(tags=["Company Routes"])
@@ -29,20 +29,34 @@ class CompanyUpdate(BaseModel):
website: Optional[str] = None
@router.get("/companies", response_model=List[CompanyData])
def read_companies(db: Session = Depends(get_db)):
"""Get all companies with their investor relationships"""
@router.get("/companies", response_model=PaginatedResponse[CompanyData])
def read_companies(
page: int = Query(1, ge=1, description="Page number (starts at 1)"),
page_size: int = Query(10, ge=1, le=100, description="Items per page (max 100)"),
db: Session = Depends(get_db),
):
"""Get all companies with their investor relationships (paginated)"""
# Calculate offset
offset = (page - 1) * page_size
# Get total count
total_count = (
db.query(CompanyTable)
.filter(CompanyTable.name.isnot(None), CompanyTable.description.isnot(None))
.count()
)
# Get paginated results
companies = (
db.query(CompanyTable)
.filter(
CompanyTable.name.isnot(None),
CompanyTable.description.isnot(None)
)
.filter(CompanyTable.name.isnot(None), CompanyTable.description.isnot(None))
.options(
selectinload(CompanyTable.investors),
selectinload(CompanyTable.members),
selectinload(CompanyTable.sectors),
)
.offset(offset)
.limit(page_size)
.all()
)
@@ -57,10 +71,19 @@ def read_companies(db: Session = Depends(get_db)):
)
company_data_list.append(company_data)
return company_data_list
# Calculate total pages
total_pages = (total_count + page_size - 1) // page_size
return PaginatedResponse(
items=company_data_list,
total=total_count,
page=page,
page_size=page_size,
total_pages=total_pages,
)
@router.get("/companies/filter", response_model=List[CompanyData])
@router.get("/companies/filter", response_model=PaginatedResponse[CompanyData])
def filter_companies(
industry: Optional[str] = Query(
None, description="Filter by industry (partial match)"
@@ -76,9 +99,11 @@ def filter_companies(
investor_name: Optional[str] = Query(
None, description="Filter by investor name (partial match)"
),
page: int = Query(1, ge=1, description="Page number (starts at 1)"),
page_size: int = Query(10, ge=1, le=100, description="Items per page (max 100)"),
db: Session = Depends(get_db),
):
"""Filter companies based on various criteria"""
"""Filter companies based on various criteria (paginated)"""
# Start with base query
query = db.query(CompanyTable).options(
@@ -112,7 +137,12 @@ def filter_companies(
InvestorTable.name.ilike(f"%{investor_name}%")
)
companies = query.all()
# Get total count before pagination
total_count = query.count()
# Calculate offset and apply pagination
offset = (page - 1) * page_size
companies = query.offset(offset).limit(page_size).all()
# Transform to CompanyData format
company_data_list = []
@@ -125,7 +155,16 @@ def filter_companies(
)
company_data_list.append(company_data)
return company_data_list
# Calculate total pages
total_pages = (total_count + page_size - 1) // page_size
return PaginatedResponse(
items=company_data_list,
total=total_count,
page=page,
page_size=page_size,
total_pages=total_pages,
)
@router.get("/companies/{company_id}", response_model=CompanyData)
app/routers/investors.py
+73 -14
View File
@@ -1,4 +1,4 @@
from typing import List, Optional
from typing import Optional
from db.db import get_db
from db.models import InvestorTable, SectorTable
@@ -8,6 +8,7 @@ from schemas.router_schemas import (
InvestmentStage,
InvestorData,
InvestorFundData,
PaginatedResponse,
)
from sqlalchemy.orm import Session, selectinload
@@ -39,13 +40,24 @@ class InvestorUpdate(BaseModel):
number_of_investments: Optional[int] = None
@router.get("/investors", response_model=List[InvestorFundData])
def read_investors(db: Session = Depends(get_db)):
"""Get all investors with their funds as separate entries
@router.get("/investors", response_model=PaginatedResponse[InvestorFundData])
def read_investors(
page: int = Query(1, ge=1, description="Page number (starts at 1)"),
page_size: int = Query(10, ge=1, le=100, description="Items per page (max 100)"),
db: Session = Depends(get_db),
):
"""Get all investors with their funds as separate entries (paginated)
Each investor-fund combination is returned as a separate row.
An investor with 3 funds will appear as 3 entries.
"""
# Calculate offset
offset = (page - 1) * page_size
# Get total count
total_count = db.query(InvestorTable).count()
# Get paginated results
investors = (
db.query(InvestorTable)
.options(
@@ -54,6 +66,8 @@ def read_investors(db: Session = Depends(get_db)):
selectinload(InvestorTable.sectors),
selectinload(InvestorTable.funds),
)
.offset(offset)
.limit(page_size)
.all()
)
@@ -124,10 +138,19 @@ def read_investors(db: Session = Depends(get_db)):
)
investor_fund_list.append(investor_fund_data)
return investor_fund_list
# Calculate total pages
total_pages = (total_count + page_size - 1) // page_size
return PaginatedResponse(
items=investor_fund_list,
total=total_count,
page=page,
page_size=page_size,
total_pages=total_pages,
)
@router.get("/investors/filter", response_model=List[InvestorFundData])
@router.get("/investors/filter", response_model=PaginatedResponse[InvestorFundData])
def filter_investors(
stage: Optional[InvestmentStage] = Query(
None, description="Filter by investment stage"
@@ -140,9 +163,11 @@ def filter_investors(
sector: Optional[str] = Query(None, description="Sector name (partial match)"),
min_aum: Optional[int] = Query(None, description="Minimum AUM"),
max_aum: Optional[int] = Query(None, description="Maximum AUM"),
page: int = Query(1, ge=1, description="Page number (starts at 1)"),
page_size: int = Query(10, ge=1, le=100, description="Items per page (max 100)"),
db: Session = Depends(get_db),
):
"""Filter investors based on various criteria
"""Filter investors based on various criteria (paginated)
Returns investor-fund combinations as separate rows.
An investor with 3 funds will appear as 3 entries.
@@ -182,7 +207,12 @@ def filter_investors(
SectorTable.name.ilike(f"%{sector}%")
)
investors = query.all()
# Get total count before pagination
total_count = query.count()
# Calculate offset and apply pagination
offset = (page - 1) * page_size
investors = query.offset(offset).limit(page_size).all()
# Transform to InvestorFundData format (one row per investor-fund combination)
investor_fund_list = []
@@ -251,7 +281,16 @@ def filter_investors(
)
investor_fund_list.append(investor_fund_data)
return investor_fund_list
# Calculate total pages
total_pages = (total_count + page_size - 1) // page_size
return PaginatedResponse(
items=investor_fund_list,
total=total_count,
page=page,
page_size=page_size,
total_pages=total_pages,
)
@router.get("/investors/{investor_id}", response_model=InvestorData)
@@ -368,13 +407,18 @@ def delete_investor(investor_id: int, db: Session = Depends(get_db)):
return {"message": "Investor deleted successfully"}
@router.get("/investors/{investor_id}/similar", response_model=List[InvestorFundData])
@router.get(
"/investors/{investor_id}/similar",
response_model=PaginatedResponse[InvestorFundData],
)
def find_similar_investors(
investor_id: int,
limit: int = Query(10, description="Maximum number of similar investors to return"),
page: int = Query(1, ge=1, description="Page number (starts at 1)"),
page_size: int = Query(10, ge=1, le=100, description="Items per page (max 100)"),
db: Session = Depends(get_db),
):
"""Find investors similar to a given investor based on characteristics
"""Find investors similar to a given investor based on characteristics (paginated)
Returns investor-fund combinations as separate rows.
"""
@@ -475,9 +519,15 @@ def find_similar_investors(
if score > 0: # Only include investors with some similarity
scored_investors.append((score, candidate))
# Sort by score (descending) and take top N
# Sort by score (descending) and take top N based on limit
scored_investors.sort(key=lambda x: x[0], reverse=True)
similar_investors = [inv for score, inv in scored_investors[:limit]]
top_similar = scored_investors[:limit]
# Apply pagination to the top similar investors
total_count = len(top_similar)
offset = (page - 1) * page_size
paginated_similar = top_similar[offset : offset + page_size]
similar_investors = [inv for score, inv in paginated_similar]
# Transform to InvestorFundData format (one row per investor-fund combination)
investor_fund_list = []
@@ -546,4 +596,13 @@ def find_similar_investors(
)
investor_fund_list.append(investor_fund_data)
return investor_fund_list
# Calculate total pages
total_pages = (total_count + page_size - 1) // page_size
return PaginatedResponse(
items=investor_fund_list,
total=total_count,
page=page,
page_size=page_size,
total_pages=total_pages,
)
app/routers/projects.py
+47 -8
View File
@@ -14,14 +14,26 @@ from schemas.project_schemas import (
ProjectData,
ProjectUpdate,
)
from schemas.router_schemas import PaginatedResponse
from sqlalchemy.orm import Session, selectinload
router = APIRouter(tags=["Project Routes"])
@router.get("/projects", response_model=List[ProjectData])
def read_projects(db: Session = Depends(get_db)):
"""Get all projects with their related data"""
@router.get("/projects", response_model=PaginatedResponse[ProjectData])
def read_projects(
page: int = Query(1, ge=1, description="Page number (starts at 1)"),
page_size: int = Query(10, ge=1, le=100, description="Items per page (max 100)"),
db: Session = Depends(get_db),
):
"""Get all projects with their related data (paginated)"""
# Calculate offset
offset = (page - 1) * page_size
# Get total count
total_count = db.query(ProjectTable).count()
# Get paginated results
projects = (
db.query(ProjectTable)
.options(
@@ -29,6 +41,8 @@ def read_projects(db: Session = Depends(get_db)):
selectinload(ProjectTable.investors),
selectinload(ProjectTable.companies),
)
.offset(offset)
.limit(page_size)
.all()
)
@@ -43,7 +57,16 @@ def read_projects(db: Session = Depends(get_db)):
)
project_data_list.append(project_data)
return project_data_list
# Calculate total pages
total_pages = (total_count + page_size - 1) // page_size
return PaginatedResponse(
items=project_data_list,
total=total_count,
page=page,
page_size=page_size,
total_pages=total_pages,
)
@router.get("/projects/{project_id}", response_model=ProjectData)
@@ -151,7 +174,7 @@ def delete_project(project_id: int, db: Session = Depends(get_db)):
return {"message": "Project deleted successfully"}
@router.get("/projects/filter", response_model=List[ProjectData])
@router.get("/projects/filter", response_model=PaginatedResponse[ProjectData])
def filter_projects(
stage: Optional[InvestmentStage] = Query(
None, description="Filter by project stage"
@@ -166,9 +189,11 @@ def filter_projects(
company_name: Optional[str] = Query(
None, description="Company name (partial match)"
),
page: int = Query(1, ge=1, description="Page number (starts at 1)"),
page_size: int = Query(10, ge=1, le=100, description="Items per page (max 100)"),
db: Session = Depends(get_db),
):
"""Filter projects based on various criteria"""
"""Filter projects based on various criteria (paginated)"""
# Start with base query
query = db.query(ProjectTable).options(
@@ -205,7 +230,12 @@ def filter_projects(
CompanyTable.name.ilike(f"%{company_name}%")
)
projects = query.all()
# Get total count before pagination
total_count = query.count()
# Calculate offset and apply pagination
offset = (page - 1) * page_size
projects = query.offset(offset).limit(page_size).all()
# Transform to ProjectData format
project_data_list = []
@@ -218,7 +248,16 @@ def filter_projects(
)
project_data_list.append(project_data)
return project_data_list
# Calculate total pages
total_pages = (total_count + page_size - 1) // page_size
return PaginatedResponse(
items=project_data_list,
total=total_count,
page=page,
page_size=page_size,
total_pages=total_pages,
)
# Association management routes