atramhasis.json_processors.provider.update_provider() - Code Metrics - Inspection of "Add Create, Update, Delete provider." - OnroerendErfgoed/atramhasis - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — develop (#812)

unknown

created 2023-03-28 12:21 UTC

update_provider() A

↳ Parent: atramhasis.json_processors.provider

Complexity

Conditions

Size

Total Lines	7
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	6
nop	3
dl	0
loc	7
rs	10
c	0
b	0
f	0

from typing import Mapping

from sqlalchemy.orm import Session

from atramhasis import mappers
from atramhasis.data.datamanagers import ProviderDataManager
from atramhasis.data.models import Provider


def create_provider(json_data: Mapping, session: Session) -> Provider:
    """Process a provider JSON into a newly stored 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)


1			from typing import Mapping
2
3			from sqlalchemy.orm import Session
4
5			from atramhasis import mappers
6			from atramhasis.data.datamanagers import ProviderDataManager
7			from atramhasis.data.models import Provider
8
9
10			def create_provider(json_data: Mapping, session: Session) -> Provider:
11			"""Process a provider JSON into a newly stored Provider."""
12			db_provider = mappers.map_provider(json_data)
13			if not db_provider.id:
14			# Store conceptscheme first so we can copy its id
15			session.add(db_provider.conceptscheme)
16			session.flush()
17			db_provider.id = str(db_provider.conceptscheme.id)
18
19			session.add(db_provider)
20			session.flush()
21
22			return db_provider
23
24
25			def update_provider(provider_id: str, json_data: Mapping, session: Session) -> Provider:
26			"""Process a JSON into to update an existing provider."""
27			manager = ProviderDataManager(session)
28			db_provider = manager.get_provider_by_id(provider_id)
29			db_provider = mappers.map_provider(json_data, provider=db_provider)
30			session.flush()
31			return db_provider
32
33
34			def delete_provider(provider_id, session: Session) -> None:
35			manager = ProviderDataManager(session)
36			db_provider = manager.get_provider_by_id(provider_id)
37			session.delete(db_provider)
38

OnroerendErfgoed / atramhasis

Pull Request — develop (#812)

update_provider() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like