CIzz22
/
be-lcca
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

add new fs table for plant transaction data

Browse Source
main
MrWaradana 4 weeks ago
parent f671249fed
commit c0f3380dbe
7 changed files with 362 additions and 0 deletions
Show Stats Download Patch File Download Diff File

BIN
src/__pycache__/api.cpython-311.pyc
View File

Binary file not shown.

7
src/api.py
Unescape Escape View File

@ -13,6 +13,7 @@ from src.masterdata_simulations.router import router as masterdata_simulations_r
from src.plant_masterdata.router import router as plant_masterdata
from src.plant_transaction_data.router import router as plant_transaction_data
from src.plant_transaction_data_simulations.router import router as plant_transaction_data_simulations
from src.plant_fs_transaction_data.router import router as plant_fs_transaction_data_router
from src.equipment.router import router as equipment_router
from src.acquisition_cost.router import router as acquisition_data_router
from src.yeardata.router import router as yeardata_router
@ -87,6 +88,12 @@ authenticated_api_router.include_router(
tags=["plant_transaction_data_simulations"],
)
authenticated_api_router.include_router(
plant_fs_transaction_data_router,
prefix="/plant-fs-transaction-data",
tags=["plant_fs_transaction_data"],
)
authenticated_api_router.include_router(
equipment_router, prefix="/equipment", tags=["equipment"]
)

1
src/plant_fs_transaction_data/__init__.py
Unescape Escape View File

@ -0,0 +1 @@
"""Plant FS transaction data module."""

29
src/plant_fs_transaction_data/model.py
Unescape Escape View File

@ -0,0 +1,29 @@
from sqlalchemy import Column, Float, Integer
from src.database.core import Base
from src.models import DefaultMixin, IdentityMixin
class PlantFSTransactionData(Base, DefaultMixin, IdentityMixin):
__tablename__ = "lcc_fs_plant_tr_data"
tahun = Column(Integer, nullable=True)
seq = Column(Integer, nullable=True)
fs_chart_total_revenue = Column(Float, nullable=True)
fs_chart_revenue_a = Column(Float, nullable=True)
fs_chart_revenue_b = Column(Float, nullable=True)
fs_chart_revenue_c = Column(Float, nullable=True)
fs_chart_revenue_d = Column(Float, nullable=True)
fs_chart_revenue_annualized = Column(Float, nullable=True)
fs_chart_fuel_cost_component_c = Column(Float, nullable=True)
fs_chart_fuel_cost = Column(Float, nullable=True)
fs_chart_fuel_cost_annualized = Column(Float, nullable=True)
fs_chart_oem_component_bd = Column(Float, nullable=True)
fs_chart_oem_bd_cost = Column(Float, nullable=True)
fs_chart_oem_periodic_maintenance_cost = Column(Float, nullable=True)
fs_chart_oem_annualized = Column(Float, nullable=True)
fs_chart_capex_component_a = Column(Float, nullable=True)
fs_chart_capex_biaya_investasi_tambahan = Column(Float, nullable=True)
fs_chart_capex_acquisition_cost = Column(Float, nullable=True)
fs_chart_capex_annualized = Column(Float, nullable=True)
fs_cost_disposal_cost = Column(Float, nullable=True)

121
src/plant_fs_transaction_data/router.py
Unescape Escape View File

@ -0,0 +1,121 @@
from typing import Optional
from fastapi import APIRouter, HTTPException, Query, status
from src.auth.service import CurrentUser
from src.database.core import DbSession
from src.database.service import CommonParameters
from src.models import StandardResponse
from .schema import (
PlantFSTransactionDataCreate,
PlantFSTransactionDataPagination,
PlantFSTransactionDataRead,
PlantFSTransactionDataUpdate,
)
from .service import create, delete, get, get_all, update
router = APIRouter()
@router.get("", response_model=StandardResponse[PlantFSTransactionDataPagination])
async def list_fs_transactions(
db_session: DbSession,
common: CommonParameters,
items_per_page: Optional[int] = Query(5),
search: Optional[str] = Query(None),
):
"""Return paginated financial statement transaction data."""
records = await get_all(
db_session=db_session,
items_per_page=items_per_page,
search=search,
common=common,
)
return StandardResponse(
data=records,
message="Data retrieved successfully",
)
@router.get(
"/{fs_transaction_id}",
response_model=StandardResponse[PlantFSTransactionDataRead],
)
async def retrieve_fs_transaction(
db_session: DbSession,
fs_transaction_id: str,
):
record = await get(db_session=db_session, fs_transaction_id=fs_transaction_id)
if not record:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="A data with this id does not exist.",
)
return StandardResponse(data=record, message="Data retrieved successfully")
@router.post("", response_model=StandardResponse[PlantFSTransactionDataRead])
async def create_fs_transaction(
db_session: DbSession,
payload: PlantFSTransactionDataCreate,
current_user: CurrentUser,
):
record = await create(
db_session=db_session,
fs_transaction_in=payload,
current_user=current_user,
)
return StandardResponse(data=record, message="Data created successfully")
@router.put(
"/{fs_transaction_id}",
response_model=StandardResponse[PlantFSTransactionDataRead],
)
async def update_fs_transaction(
db_session: DbSession,
fs_transaction_id: str,
payload: PlantFSTransactionDataUpdate,
current_user: CurrentUser,
):
record = await get(db_session=db_session, fs_transaction_id=fs_transaction_id)
if not record:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="A data with this id does not exist.",
)
updated = await update(
db_session=db_session,
fs_transaction=record,
fs_transaction_in=payload,
current_user=current_user,
)
return StandardResponse(data=updated, message="Data updated successfully")
@router.delete(
"/{fs_transaction_id}",
response_model=StandardResponse[PlantFSTransactionDataRead],
)
async def delete_fs_transaction(
db_session: DbSession,
fs_transaction_id: str,
):
record = await get(db_session=db_session, fs_transaction_id=fs_transaction_id)
if not record:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=[{"msg": "A data with this id does not exist."}],
)
await delete(db_session=db_session, fs_transaction_id=fs_transaction_id)
return StandardResponse(data=record, message="Data deleted successfully")

86
src/plant_fs_transaction_data/schema.py
Unescape Escape View File

@ -0,0 +1,86 @@
from datetime import datetime
from typing import List, Optional
from uuid import UUID
from pydantic import Field
from src.models import DefaultBase, Pagination
class PlantFSTransactionDataBase(DefaultBase):
fs_chart_total_revenue: Optional[float] = Field(
None, nullable=True, ge=0, le=1_000_000_000_000_000
)
fs_chart_revenue_a: Optional[float] = Field(
None, nullable=True, ge=0, le=1_000_000_000_000_000
)
fs_chart_revenue_b: Optional[float] = Field(
None, nullable=True, ge=0, le=1_000_000_000_000_000
)
fs_chart_revenue_c: Optional[float] = Field(
None, nullable=True, ge=0, le=1_000_000_000_000_000
)
fs_chart_revenue_d: Optional[float] = Field(
None, nullable=True, ge=0, le=1_000_000_000_000_000
)
fs_chart_revenue_annualized: Optional[float] = Field(
None, nullable=True, ge=0, le=1_000_000_000_000_000
)
fs_chart_fuel_cost_component_c: Optional[float] = Field(
None, nullable=True, ge=0, le=1_000_000_000_000_000
)
fs_chart_fuel_cost: Optional[float] = Field(
None, nullable=True, ge=0, le=1_000_000_000_000_000
)
fs_chart_fuel_cost_annualized: Optional[float] = Field(
None, nullable=True, ge=0, le=1_000_000_000_000_000
)
fs_chart_oem_component_bd: Optional[float] = Field(
None, nullable=True, ge=0, le=1_000_000_000_000_000
)
fs_chart_oem_bd_cost: Optional[float] = Field(
None, nullable=True, ge=0, le=1_000_000_000_000_000
)
fs_chart_oem_periodic_maintenance_cost: Optional[float] = Field(
None, nullable=True, ge=0, le=1_000_000_000_000_000
)
fs_chart_oem_annualized: Optional[float] = Field(
None, nullable=True, ge=0, le=1_000_000_000_000_000
)
fs_chart_capex_component_a: Optional[float] = Field(
None, nullable=True, ge=0, le=1_000_000_000_000_000
)
fs_chart_capex_biaya_investasi_tambahan: Optional[float] = Field(
None, nullable=True, ge=0, le=1_000_000_000_000_000
)
fs_chart_capex_acquisition_cost: Optional[float] = Field(
None, nullable=True, ge=0, le=1_000_000_000_000_000
)
fs_chart_capex_annualized: Optional[float] = Field(
None, nullable=True, ge=0, le=1_000_000_000_000_000
)
fs_cost_disposal_cost: Optional[float] = Field(
None, nullable=True, ge=0, le=1_000_000_000_000_000
)
tahun: Optional[int] = Field(None, nullable=True, ge=1900, le=9999)
seq: Optional[int] = Field(None, nullable=True, ge=0, le=9999)
created_at: Optional[datetime] = Field(None, nullable=True)
updated_at: Optional[datetime] = Field(None, nullable=True)
created_by: Optional[str] = Field(None, nullable=True)
updated_by: Optional[str] = Field(None, nullable=True)
class PlantFSTransactionDataCreate(PlantFSTransactionDataBase):
pass
class PlantFSTransactionDataUpdate(PlantFSTransactionDataBase):
pass
class PlantFSTransactionDataRead(PlantFSTransactionDataBase):
id: UUID
class PlantFSTransactionDataPagination(Pagination):
items: List[PlantFSTransactionDataRead] = []

118
src/plant_fs_transaction_data/service.py
Unescape Escape View File

@ -0,0 +1,118 @@
import logging
from typing import Optional
from sqlalchemy import Delete, Select, String, cast
from src.auth.service import CurrentUser
from src.database.core import DbSession
from src.database.service import search_filter_sort_paginate
from src.plant_fs_transaction_data.model import PlantFSTransactionData
from src.plant_fs_transaction_data.schema import (
PlantFSTransactionDataCreate,
PlantFSTransactionDataUpdate,
)
logger = logging.getLogger(__name__)
def _resolve_user_id(user: Optional[CurrentUser]) -> Optional[str]:
"""Return the current user's identifier when available."""
if user is None:
return None
# UserBase guarantees user_id
return getattr(user, "user_id", None)
async def get(
*, db_session: DbSession, fs_transaction_id: str
) -> Optional[PlantFSTransactionData]:
"""Return a record by its primary key."""
query = Select(PlantFSTransactionData).where(
PlantFSTransactionData.id == fs_transaction_id
)
result = await db_session.execute(query)
return result.scalars().one_or_none()
def _apply_text_search(query: Select, search: Optional[str]):
if not search:
return query
wildcard = f"%{search}%"
return query.where(
cast(PlantFSTransactionData.tahun, String).ilike(wildcard)
| cast(PlantFSTransactionData.seq, String).ilike(wildcard)
| PlantFSTransactionData.created_by.ilike(wildcard)
)
async def get_all(
*,
db_session: DbSession,
items_per_page: Optional[int],
search: Optional[str] = None,
common,
):
"""Return paginated FS transaction data."""
query = (
Select(PlantFSTransactionData)
.order_by(
PlantFSTransactionData.seq.asc(),
PlantFSTransactionData.tahun.asc(),
)
)
query = _apply_text_search(query, search)
common["items_per_page"] = items_per_page
return await search_filter_sort_paginate(model=query, **common)
async def create(
*,
db_session: DbSession,
fs_transaction_in: PlantFSTransactionDataCreate,
current_user: Optional[CurrentUser] = None,
) -> PlantFSTransactionData:
"""Create a new FS record."""
payload = fs_transaction_in.model_dump()
user_id = _resolve_user_id(current_user)
if user_id:
payload.setdefault("created_by", user_id)
payload.setdefault("updated_by", user_id)
fs_transaction = PlantFSTransactionData(**payload)
db_session.add(fs_transaction)
await db_session.commit()
await db_session.refresh(fs_transaction)
return fs_transaction
async def update(
*,
db_session: DbSession,
fs_transaction: PlantFSTransactionData,
fs_transaction_in: PlantFSTransactionDataUpdate,
current_user: Optional[CurrentUser] = None,
) -> PlantFSTransactionData:
"""Update an existing FS record."""
update_data = fs_transaction_in.model_dump(exclude_defaults=True)
user_id = _resolve_user_id(current_user)
if user_id:
update_data.setdefault("updated_by", user_id)
for field, value in update_data.items():
setattr(fs_transaction, field, value)
await db_session.commit()
await db_session.refresh(fs_transaction)
return fs_transaction
async def delete(*, db_session: DbSession, fs_transaction_id: str) -> None:
"""Delete a record."""
query = Delete(PlantFSTransactionData).where(
PlantFSTransactionData.id == fs_transaction_id
)
await db_session.execute(query)
await db_session.commit()
Write Preview
Loading…
Cancel
Save