30 lines
1.0 KiB
Python
30 lines
1.0 KiB
Python
import datetime
|
|
from typing import Union
|
|
|
|
from sqlalchemy import select
|
|
|
|
from models import Client
|
|
from services.base import BaseService
|
|
from schemas.client import *
|
|
|
|
|
|
class ClientService(BaseService):
|
|
|
|
async def get_by_name(self, name: str) -> Union[Client, None]:
|
|
client = await self.session.scalar(select(Client).where(Client.name == name))
|
|
return client
|
|
|
|
async def create_client_raw(self, name: str, address: str) -> Client:
|
|
client = Client(name=name, address=address, created_at=datetime.datetime.now())
|
|
self.session.add(client)
|
|
await self.session.flush()
|
|
return client
|
|
|
|
async def search_clients(self, request: ClientSearchRequest) -> ClientSearchResponse:
|
|
query = await self.session.scalars(select(Client)
|
|
.where(Client.name.ilike(f'%{request.name}%')))
|
|
clients = []
|
|
for client in query.all():
|
|
clients.append(ClientSchema.model_validate(client))
|
|
return ClientSearchResponse(clients=clients)
|