656f596d7e
Core modules: - Laws: CRUD, search, AI-powered QA - Analysis: legal research and case management - Contracts: lifecycle management with templates - Signatures: electronic signature workflow Infrastructure: - FastAPI + SQLite + async SQLAlchemy - Docker deployment support - 54 unit tests passing Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
138 lines
3.8 KiB
Python
138 lines
3.8 KiB
Python
"""Law service for CRUD operations."""
|
|
from datetime import date
|
|
from typing import List, Optional
|
|
from sqlalchemy import select, or_
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.law import Law, LawArticle, LawType, LawStatus
|
|
|
|
|
|
class LawService:
|
|
"""Service for law CRUD operations."""
|
|
|
|
def __init__(self, db: AsyncSession):
|
|
self.db = db
|
|
|
|
async def create_law(
|
|
self,
|
|
title: str,
|
|
law_type: LawType,
|
|
promulgation_date: date,
|
|
effective_date: date,
|
|
issuing_authority: str,
|
|
content: str,
|
|
status: LawStatus = LawStatus.EFFECTIVE,
|
|
document_number: Optional[str] = None,
|
|
) -> Law:
|
|
"""Create a new law."""
|
|
law = Law(
|
|
title=title,
|
|
law_type=law_type,
|
|
promulgation_date=promulgation_date,
|
|
effective_date=effective_date,
|
|
status=status,
|
|
issuing_authority=issuing_authority,
|
|
content=content,
|
|
document_number=document_number,
|
|
)
|
|
self.db.add(law)
|
|
await self.db.flush()
|
|
await self.db.refresh(law)
|
|
return law
|
|
|
|
async def get_law_by_id(self, law_id: int) -> Optional[Law]:
|
|
"""Get a law by ID."""
|
|
result = await self.db.execute(
|
|
select(Law).where(Law.id == law_id)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def get_laws_list(
|
|
self,
|
|
skip: int = 0,
|
|
limit: int = 20,
|
|
law_type: Optional[LawType] = None,
|
|
status: Optional[LawStatus] = None,
|
|
) -> List[Law]:
|
|
"""Get list of laws with optional filters."""
|
|
query = select(Law)
|
|
|
|
if law_type:
|
|
query = query.where(Law.law_type == law_type)
|
|
if status:
|
|
query = query.where(Law.status == status)
|
|
|
|
query = query.offset(skip).limit(limit).order_by(Law.created_at.desc())
|
|
result = await self.db.execute(query)
|
|
return list(result.scalars().all())
|
|
|
|
async def update_law(
|
|
self,
|
|
law_id: int,
|
|
**kwargs
|
|
) -> Optional[Law]:
|
|
"""Update a law."""
|
|
law = await self.get_law_by_id(law_id)
|
|
if not law:
|
|
return None
|
|
|
|
for key, value in kwargs.items():
|
|
if hasattr(law, key) and value is not None:
|
|
setattr(law, key, value)
|
|
|
|
await self.db.flush()
|
|
await self.db.refresh(law)
|
|
return law
|
|
|
|
async def delete_law(self, law_id: int) -> bool:
|
|
"""Delete a law."""
|
|
law = await self.get_law_by_id(law_id)
|
|
if not law:
|
|
return False
|
|
|
|
await self.db.delete(law)
|
|
await self.db.flush()
|
|
return True
|
|
|
|
async def search_laws_by_keyword(
|
|
self,
|
|
keyword: str,
|
|
limit: int = 10
|
|
) -> List[Law]:
|
|
"""Search laws by keyword in title or content."""
|
|
query = select(Law).where(
|
|
or_(
|
|
Law.title.contains(keyword),
|
|
Law.content.contains(keyword),
|
|
)
|
|
).limit(limit)
|
|
|
|
result = await self.db.execute(query)
|
|
return list(result.scalars().all())
|
|
|
|
async def create_article(
|
|
self,
|
|
law_id: int,
|
|
article_number: str,
|
|
content: str,
|
|
) -> LawArticle:
|
|
"""Create a law article."""
|
|
article = LawArticle(
|
|
law_id=law_id,
|
|
article_number=article_number,
|
|
content=content,
|
|
)
|
|
self.db.add(article)
|
|
await self.db.flush()
|
|
await self.db.refresh(article)
|
|
return article
|
|
|
|
async def get_articles_by_law_id(self, law_id: int) -> List[LawArticle]:
|
|
"""Get articles for a law."""
|
|
result = await self.db.execute(
|
|
select(LawArticle)
|
|
.where(LawArticle.law_id == law_id)
|
|
.order_by(LawArticle.id)
|
|
)
|
|
return list(result.scalars().all())
|