Skip to content
Extraits de code Groupes Projets

Added fastapi framework

4 files
+ 173
176
Comparer les modifications
  • Côte à côte
  • En ligne

Fichiers

+ 1
175
import hashlib
import json
from typing import Annotated
from fastapi import APIRouter, Depends
from odoo import fields, models
from odoo.api import Environment
from odoo.addons.fastapi.dependencies import odoo_env
from ..schemas import Event, Training
from ..routers.cgscop_event_api import cg_event_training_router
class TrainingsEndpoint(models.Model):
@@ -24,168 +15,3 @@ class TrainingsEndpoint(models.Model):
if self.app == "cg_event_training":
return [cg_event_training_router]
return super()._get_fastapi_routers()
# create a router
cg_event_training_router = APIRouter()
@cg_event_training_router.get("/trainings", response_model=list[Training])
def get_trainings(env: Annotated[Environment, Depends(odoo_env)]) -> list[Training]:
domain = []
trainings = []
for t in env["event_training"].sudo().search(domain):
list_events = []
# Does event needs a domain ? Like taking only published ones or we take all linked
for e in t.event_ids:
e_website = (
(e.get_base_url() + e.website_url)
if e.event_privacy == "public"
else ""
)
# Every attribute is str() casted to avoid conflict with the schema when a field
# is empty.
# Empty char field for instance return False, where the schema expect a string
current_event = Event(
title=str(e.name),
category="To be defined",
date_begin=fields.Datetime.to_string(e.date_begin),
date_end=fields.Datetime.to_string(e.date_end),
description=str(e.description),
attachment="To be defined",
horaire=fields.Datetime.to_string(e.date_begin),
address=str(e.address_id.name),
fax="To be defined",
website_url=str(e_website),
theme="To be defined",
city=str(e.address_id.city),
region="To be defined",
state=str(e.stage_id.name),
)
list_events.append(current_event)
# Every attribute is str() casted to avoid conflict with the schema when a field
# is empty.
# Empty char field for instance return False, where the schema expect a string
trainings.append(
Training(
title=str(t.name),
theme=str("To be defined"),
is_subscriber=t.is_subscriber,
text_body=str(t.text),
description=str(str(t.description)),
other_training=str(str(t.other_training)),
modalities=str(t.modalities),
pre_requisite=str(t.pre_requisite),
duration=str("To be defined"),
other_infos=str(t.other_infos),
price=str(t.price),
certification=str(t.certification),
attachment=str(t.attachment),
formation_type=str(t.formation_type),
events=list_events,
)
)
return trainings
@cg_event_training_router.get("/trainings/checksum", response_model=dict)
def get_trainings_checksum(env: Annotated[Environment, Depends(odoo_env)]) -> dict:
e_domain = [
("training_id", "!=", False),
]
t_domain = []
events_with_trainings = env["event.event"].sudo().search(e_domain)
trainings = env["event_training"].sudo().search(t_domain)
data = {}
for t in trainings.read():
data = {**data, **t}
for e in events_with_trainings.read():
data = {**data, **e}
encoded_data = json.dumps(data, sort_keys=True, default=str).encode()
h = hashlib.sha256()
h.update(encoded_data)
checksum_dict = {"checksum": h.hexdigest()}
return checksum_dict
@cg_event_training_router.get("/events", response_model=list[Event])
def get_events(env: Annotated[Environment, Depends(odoo_env)]) -> list[Event]:
domain = [
("is_published", "=", True),
]
list_events = []
for e in env["event.event"].sudo().search(domain):
e_website = (
(e.get_base_url() + e.website_url) if e.event_privacy == "public" else ""
)
# Every attribute is str() casted to avoid conflict with the schema when a field
# is empty.
# Empty char field for instance return False, where the schema expect a string
current_event = Event(
title=str(e.name),
category="To be defined",
date_begin=fields.Datetime.to_string(e.date_begin),
date_end=fields.Datetime.to_string(e.date_end),
description=str(e.description),
attachment="To be defined",
horaire=fields.Datetime.to_string(e.date_begin),
address=str(e.address_id.name),
fax="To be defined",
website_url=str(e_website),
theme="To be defined",
city=str(e.address_id.city),
region="To be defined",
state=str(e.stage_id.name),
)
list_events.append(current_event)
return list_events
@cg_event_training_router.get("/event/{event_id}", response_model=list[Event])
def get_event_by_id(
env: Annotated[Environment, Depends(odoo_env)], event_id: int
) -> list[Event]:
list_events = []
for e in env["event.event"].sudo().browse([event_id]):
e_website = (
(e.get_base_url() + e.website_url) if e.event_privacy == "public" else ""
)
# Every attribute is str() casted to avoid conflict with the schema when a field
# is empty.
# Empty char field for instance return False, where the schema expect a string
current_event = Event(
title=str(e.name),
category="To be defined",
date_begin=fields.Datetime.to_string(e.date_begin),
date_end=fields.Datetime.to_string(e.date_end),
description=str(e.description),
attachment="To be defined",
horaire=fields.Datetime.to_string(e.date_begin),
address=str(e.address_id.name),
fax="To be defined",
website_url=str(e_website),
theme="To be defined",
city=str(e.address_id.city),
region="To be defined",
state=str(e.stage_id.name),
)
list_events.append(current_event)
return list_events
# @demo_api_router.get("/partners")
# def get_partners(env: Annotated[Environment, Depends(odoo_env)]):
# search_ = [p for p in env["res.partner"].sudo().search([])]
# return search_
Chargement en cours