forked from Netflix/dispatch
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmodels.py
More file actions
106 lines (85 loc) · 3.54 KB
/
Copy pathmodels.py
File metadata and controls
106 lines (85 loc) · 3.54 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
from datetime import datetime
from typing import List, Optional
from sqlalchemy import Boolean, Column, ForeignKey, Integer, PrimaryKeyConstraint, String, Table
from sqlalchemy.orm import backref, relationship
from sqlalchemy_utils import TSVectorType
from dispatch.database import Base
from dispatch.incident_priority.models import IncidentPriorityCreate, IncidentPriorityRead
from dispatch.incident_type.models import IncidentTypeCreate, IncidentTypeRead
from dispatch.models import DispatchBase, TermReadNested, TimeStampMixin
from dispatch.term.models import TermCreate
# Association tables for many to many relationships
assoc_service_incident_priorities = Table(
"service_incident_priority",
Base.metadata,
Column("incident_priority_id", Integer, ForeignKey("incident_priority.id")),
Column("service_id", Integer, ForeignKey("service.id")),
PrimaryKeyConstraint("incident_priority_id", "service_id"),
)
assoc_service_incident_types = Table(
"service_incident_type",
Base.metadata,
Column("incident_type_id", Integer, ForeignKey("incident_type.id")),
Column("service_id", Integer, ForeignKey("service.id")),
PrimaryKeyConstraint("incident_type_id", "service_id"),
)
assoc_service_incidents = Table(
"service_incident",
Base.metadata,
Column("incident_id", Integer, ForeignKey("incident.id")),
Column("service_id", Integer, ForeignKey("service.id")),
PrimaryKeyConstraint("incident_id", "service_id"),
)
assoc_service_terms = Table(
"service_terms",
Base.metadata,
Column("term_id", Integer, ForeignKey("term.id")),
Column("service_id", Integer, ForeignKey("service.id")),
PrimaryKeyConstraint("term_id", "service_id"),
)
# SQLAlchemy models...
class Service(TimeStampMixin, Base):
id = Column(Integer, primary_key=True)
is_active = Column(Boolean, default=True)
name = Column(String)
type = Column(String, default="pagerduty-oncall")
description = Column(String)
external_id = Column(String)
incidents = relationship("Incident", secondary=assoc_service_incidents, backref="services")
incident_priorities = relationship(
"IncidentPriority", secondary=assoc_service_incident_priorities, backref="services"
)
incident_types = relationship(
"IncidentType", secondary=assoc_service_incident_types, backref="services"
)
terms = relationship(
"Term", secondary=assoc_service_terms, backref=backref("services", cascade="all")
)
search_vector = Column(TSVectorType("name"))
# Pydantic models...
class ServiceBase(DispatchBase):
name: Optional[str] = None
external_id: Optional[str] = None
description: Optional[str] = None
is_active: Optional[bool] = None
type: Optional[str] = None
class ServiceCreate(ServiceBase):
terms: Optional[List[TermCreate]] = []
incident_priorities: Optional[List[IncidentPriorityCreate]] = []
incident_types: Optional[List[IncidentTypeCreate]] = []
class ServiceUpdate(ServiceBase):
terms: Optional[List[TermCreate]] = []
incident_priorities: Optional[List[IncidentPriorityCreate]] = []
incident_types: Optional[List[IncidentTypeCreate]] = []
class ServiceRead(ServiceBase):
id: int
incident_priorities: Optional[List[IncidentPriorityRead]] = []
incident_types: Optional[List[IncidentTypeRead]] = []
terms: Optional[List[TermReadNested]] = []
created_at: Optional[datetime] = None
updated_at: Optional[datetime] = None
class ServiceNested(ServiceBase):
id: int
class ServicePagination(DispatchBase):
total: int
items: List[ServiceRead] = []