-
Notifications
You must be signed in to change notification settings - Fork 11
/
provider.py
45 lines (35 loc) · 1.62 KB
/
provider.py
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
from typing import Mapping
from skosprovider.registry import Registry
from sqlalchemy.orm import Session
from atramhasis import mappers
from atramhasis.data.datamanagers import ProviderDataManager
from atramhasis.data.models import Provider
from atramhasis.errors import ValidationError
def create_provider(json_data: Mapping, session: Session, skos_registry: Registry) -> Provider:
"""Process a provider JSON into a newly stored Provider."""
for provider in skos_registry.get_providers():
if provider.get_vocabulary_uri() == json_data["conceptscheme_uri"]:
raise ValidationError(
"Provider could not be validated.",
[{"conceptscheme_uri": "Collides with existing provider."}],
)
db_provider = mappers.map_provider(json_data)
if not db_provider.id:
# Store conceptscheme first so we can copy its id
session.add(db_provider.conceptscheme)
session.flush()
db_provider.id = str(db_provider.conceptscheme.id)
session.add(db_provider)
session.flush()
return db_provider
def update_provider(provider_id: str, json_data: Mapping, session: Session) -> Provider:
"""Process a JSON into to update an existing provider."""
manager = ProviderDataManager(session)
db_provider = manager.get_provider_by_id(provider_id)
db_provider = mappers.map_provider(json_data, provider=db_provider)
session.flush()
return db_provider
def delete_provider(provider_id, session: Session) -> None:
manager = ProviderDataManager(session)
db_provider = manager.get_provider_by_id(provider_id)
session.delete(db_provider)