"""Logic for ProjectTransferTerm endpoints.""" from abacus_common_logic.connectors.database import db from abacus_common_logic.utils.users import get_flask_user_id from owsresponse import response from abacus_contract.logic.contract_term_transfer import create_transfer_contract_term from abacus_contract.models.contract_term import ContractTerm from abacus_contract.models.contract_term_condition import ContractTermCondition from royalties.models import ProjectTransferTerm, ProjectTransferTermCondition from royalties.schemas import ( ProjectTransferTermConditionDetailSchema, ProjectTransferTermDetailSchema, ProjectTransferTermInputSchema, ) def get_terms_for_job(job_id: int) -> response.Response: """Return all transfer terms for a job, with their nested conditions. Args: job_id: id of the project transfer job. Returns: ows Response with a JSON array of terms; empty array if the job has no terms. """ terms = ProjectTransferTerm.query.filter(ProjectTransferTerm.job_id == job_id).all() conditions_by_term = _conditions_by_term_id( [t.project_transfer_term_id for t in terms] ) payload = [ _serialize_term(term, conditions_by_term.get(term.project_transfer_term_id, [])) for term in terms ] return response.Response(message=payload, status=200) def create_terms_for_job(job_id: int, terms_input: list) -> response.Response: """Create transfer terms (and their conditions) for a job. Persists each term + its conditions in a single transaction. On success, returns the freshly created terms in the same response shape as the GET. A label input with no conditions means "transfer into the destination contract's label terms as they are": it is expanded into one staged row per active label contract_term, mirroring the term's fields, snapshotting its conditions, and pre-linking both destination id columns (the destination rows already exist, so there is nothing for the transfer execution to create). Args: job_id: id of the project transfer job to attach the terms to. terms_input: list of `ProjectTransferTermInputSchema`-shaped dicts. Returns: ows Response with the created terms (status 201). """ if not terms_input: return response.Response(message=[], status=201) schema = ProjectTransferTermInputSchema(many=True) validated = schema.load(terms_input) created_by = str(get_flask_user_id()) now = ProjectTransferTerm.current_timestamp() try: specs = [] for item in validated: specs.extend(_term_specs_for_input(item)) terms = [ ProjectTransferTerm.build( job_id=job_id, created_by=created_by, created_at=now, **term_kwargs, ) for term_kwargs, _ in specs ] # Single flush populates id for every term. db.session.flush() created_terms = [] for term, (_, condition_kwargs) in zip(terms, specs): conditions = [ ProjectTransferTermCondition.build( project_transfer_term_id=term.project_transfer_term_id, created_by=created_by, created_at=now, **kwargs, ) for kwargs in condition_kwargs ] created_terms.append((term, conditions)) db.session.commit() except Exception: db.session.rollback() raise payload = [_serialize_term(term, conds) for term, conds in created_terms] return response.Response(message=payload, status=201) def _term_specs_for_input(item: dict) -> list: """Expand one validated input item into (term_kwargs, condition_kwargs) pairs. A bare label input (no conditions) expands to one spec per active label contract_term on the contract; anything else maps to a single spec carrying the input as sent. Falls back to the as-sent spec when the contract has no active label terms. """ if item['term_type'] == 'label' and not item['conditions']: label_specs = _label_term_specs(item['contract_id']) if label_specs: return label_specs return [ ( { 'contract_id': item['contract_id'], 'name': item.get('name'), 'term_type': item['term_type'], 'attachments': item.get('attachments'), 'attachment_relations': item.get('attachment_relations'), }, [ { 'name': cond.get('name'), 'priority': cond['priority'], 'term_rate': cond['term_rate'], 'commission': cond.get('commission', 0), 'conditions': cond.get('conditions') or {}, } for cond in item['conditions'] ], ) ] def _label_term_specs(contract_id: int) -> list: """Build staged-term specs mirroring each active label term on a contract.""" label_terms = ( ContractTerm.query.filter( ContractTerm.contract_id == contract_id, ContractTerm.term_type == 'label', ContractTerm.deleted_at.is_(None), ) .order_by(ContractTerm.contract_term_id) .all() ) specs = [] for label_term in label_terms: snapshots = ContractTermCondition.get_active_term_conditions_with_order_by( label_term.contract_term_id ) specs.append( ( { 'contract_id': contract_id, 'name': label_term.contract_term_name, 'term_type': 'label', 'attachments': label_term.attachments, 'attachment_relations': label_term.attachments_relations, 'destination_contract_term_id': label_term.contract_term_id, }, [ { 'name': cond.contract_term_condition_name, 'priority': cond.priority, 'term_rate': cond.term_rate, 'commission': cond.commission, 'conditions': cond.conditions or {}, 'destination_contract_term_condition_id': ( cond.contract_term_condition_id ), } for cond in snapshots ], ) ) return specs def set_destination_term_id( term_id: int, destination_contract_term_id: int ) -> response.Response: """Set destination_contract_term_id on a staged transfer term after the lambda runs.""" term = ProjectTransferTerm.query.get(term_id) if term is None: return response.create_error_response( code='not_found', message=f'Transfer term {term_id} not found.', status=404, ) try: term.update_attributes( destination_contract_term_id=destination_contract_term_id ) db.session.commit() except Exception: db.session.rollback() raise conditions = _conditions_by_term_id([term.project_transfer_term_id]).get( term.project_transfer_term_id, [] ) return response.Response(message=_serialize_term(term, conditions), status=200) def create_destination_term( project_transfer_term_id: int, contract_id: int, term_type: str, attachments: list, conditions: list, attachment_relations: dict = None, name: str = None, ) -> response.Response: """Idempotently create the destination contract_term for a staged transfer term. Keyed on project_transfer_term_id: if the staged term already has a destination_contract_term_id (a prior call already created the term), return that existing term with status 200 instead of creating a duplicate. Otherwise create the contract_term and record the links (term + each condition) on the staged rows in the SAME transaction, so a retry can never produce two contract_terms for one staged term — even if the caller dies right after. The staged term row is locked FOR UPDATE so concurrent calls for the same project_transfer_term_id serialize: the second waits, then sees the link set and takes the idempotent path instead of creating a second contract_term. Returns 201 on create, 200 on idempotent hit, 404 if the staged term is gone. """ term = ( ProjectTransferTerm.query.with_for_update() .filter( ProjectTransferTerm.project_transfer_term_id == project_transfer_term_id ) .first() ) if term is None: return response.create_error_response( code='not_found', message=f'Transfer term {project_transfer_term_id} not found.', status=404, ) if term.destination_contract_term_id is not None: return response.Response( message={ 'contract_term_id': term.destination_contract_term_id, 'contract_id': contract_id, 'term_type': term_type, 'idempotent': True, }, status=200, ) try: # commit=False keeps the transaction open so the contract_term, its # conditions, and the destination links below all commit atomically. result = create_transfer_contract_term( contract_id=contract_id, term_type=term_type, attachments=attachments, conditions=conditions, attachment_relations=attachment_relations, name=name, commit=False, ) term.update_attributes(destination_contract_term_id=result['contract_term_id']) _link_destination_conditions( project_transfer_term_id, result.get('conditions', []) ) db.session.commit() except Exception: db.session.rollback() raise return response.Response(message=result, status=201) def _link_destination_conditions( project_transfer_term_id: int, created_conditions: list ) -> None: """Record destination_contract_term_condition_id on each staged condition. Matches each staged project_transfer_term_condition to the contract_term_condition just created for it by priority (the same key the backfill uses). Leaves a staged condition unlinked if no created condition shares its priority. """ id_by_priority = { c['priority']: c['contract_term_condition_id'] for c in created_conditions } staged_conditions = ProjectTransferTermCondition.query.filter( ProjectTransferTermCondition.project_transfer_term_id == project_transfer_term_id ).all() for staged in staged_conditions: dest_id = id_by_priority.get(staged.priority) if dest_id is not None: staged.update_attributes(destination_contract_term_condition_id=dest_id) def get_active_label_term_for_contract(contract_id: int) -> response.Response: """Return the active label contract_term for a contract. Used by the AccountingTransfer lambda to resolve the current label term at execution time rather than relying on a value set at staging time. """ term = ( ContractTerm.query.filter(ContractTerm.contract_id == contract_id) .filter(ContractTerm.term_type == 'label') .filter(ContractTerm.deleted_at.is_(None)) .order_by(ContractTerm.contract_term_id.desc()) .first() ) if term is None: return response.create_error_response( code='not_found', message=f'No active label term found for contract {contract_id}.', status=404, ) return response.Response( message={ 'contract_term_id': term.contract_term_id, 'contract_id': term.contract_id, }, status=200, ) def _conditions_by_term_id(term_ids: list) -> dict: """Fetch all conditions for a set of term ids, grouped by term id.""" if not term_ids: return {} rows = ProjectTransferTermCondition.query.filter( ProjectTransferTermCondition.project_transfer_term_id.in_(term_ids) ).all() grouped = {} for row in rows: grouped.setdefault(row.project_transfer_term_id, []).append(row) return grouped def _serialize_term(term, conditions): """Serialize a term and its conditions into the API response shape.""" schema = ProjectTransferTermDetailSchema() payload = schema.dump(term) payload['conditions'] = [_serialize_condition(c) for c in conditions] return payload def _serialize_condition(condition): """Serialize a single condition row.""" return ProjectTransferTermConditionDetailSchema().dump(condition)