from onegov.core.orm import Base
from onegov.core.orm.mixins import ContentMixin
from onegov.core.orm.mixins import TimestampMixin
from onegov.pay.models.payment import Payment
from sqlalchemy import column
from sqlalchemy import Index
from sqlalchemy.orm import mapped_column, relationship, Mapped
from uuid import uuid4, UUID
from typing import Any, TYPE_CHECKING
if TYPE_CHECKING:
# we are shadowing type below
from builtins import type as _type
from collections.abc import Mapping
from decimal import Decimal
from markupsafe import Markup
from onegov.core.request import CoreRequest
from onegov.pay import Price
from onegov.pay.types import PaymentState
[docs]
class PaymentProvider[PaymentT: Payment = Payment](
Base, TimestampMixin, ContentMixin
):
""" Represents a payment provider. """
[docs]
__tablename__ = 'payment_providers'
#: the public id of the payment provider
[docs]
id: Mapped[UUID] = mapped_column(
primary_key=True,
default=uuid4
)
#: the polymorphic type of the provider
[docs]
type: Mapped[str] = mapped_column(default=lambda: 'generic')
#: true if this is the default provider (can only ever be one)
[docs]
default: Mapped[bool] = mapped_column(default=False)
#: true if this provider is enabled
[docs]
enabled: Mapped[bool] = mapped_column(default=True)
[docs]
__mapper_args__ = {
'polymorphic_on': type,
'polymorphic_identity': 'generic'
}
[docs]
__table_args__ = (
Index(
'only_one_default_provider', 'default',
unique=True, postgresql_where=column('default') == True
),
)
[docs]
payments: Mapped[list[Payment]] = relationship(
order_by='Payment.created',
back_populates='provider',
passive_deletes=True
)
if TYPE_CHECKING:
@property
[docs]
def payment_class(self) -> _type[PaymentT]: ...
else:
@property
def payment_class(self) -> _type[Payment]:
assert type(self) is PaymentProvider, 'Override this in subclasses'
return Payment
[docs]
def payment(
self,
*,
amount: Decimal | None = None,
currency: str = 'CHF',
remote_id: str | None = None,
state: PaymentState = 'open',
# FIXME: We probably don't want to allow arbitrary kwargs
# but we need to make sure, we don't use any other
# one somewhere first
**kwargs: Any
) -> PaymentT:
""" Creates a new payment using the correct model. """
payment = self.payment_class(
amount=amount,
currency=currency,
remote_id=remote_id,
state=state,
**kwargs
)
payment.provider = self
return payment
[docs]
def adjust_price(self, price: Price | None) -> Price | None:
""" Called by client implementations this method allows to adjust the
price by adding a fee to it.
By default no change is made.
"""
return price
[docs]
def charge(self, amount: Decimal, currency: str, token: str) -> Payment:
""" Given a payment token, charges the customer and creates a payment
which is returned.
"""
raise NotImplementedError
@property
[docs]
def title(self) -> str:
""" The title of the payment provider (i.e. the product name). """
raise NotImplementedError
@property
[docs]
def url(self) -> str:
""" The url to the backend of the payment provider. """
raise NotImplementedError
@property
[docs]
def public_identity(self) -> str:
""" The public identifier of this payment provider. For example, the
account name.
"""
raise NotImplementedError
@property
[docs]
def identity(self) -> str | None:
""" Uniquely identifies this payment provider amongst other providers
of the same type (say the private key of the api). Used to be able
to tell if a new oauth connection is the same as an existing one.
This identity is not meant to be displayed.
"""
raise NotImplementedError
@property
[docs]
def connected(self) -> bool:
""" True if the provider is properly hooked up to the payment
provider.
"""
return False
[docs]
def get_token(self, request: CoreRequest) -> str | None:
""" Extracts this provider's specific token from the request. """
raise NotImplementedError
[docs]
def prepare_oauth_request(
self,
redirect_url: str,
success_url: str,
error_url: str,
user_fields: dict[str, Any] | None = None
) -> str:
""" Registers the oauth request with the oauth_gateway and returns
an url that is ready to be used for the complete oauth request.
"""
raise NotImplementedError
[docs]
def process_oauth_response(
self,
request_params: Mapping[str, Any]
) -> None:
""" Processes the oauth response using the parameters passed by
the returning oauth request via the gateway.
"""
raise NotImplementedError
[docs]
def sync(self) -> None:
""" Updates the local payment information with the information from
the remote payment provider.
"""
@property
[docs]
def payment_via_get(self) -> bool:
"""
Whether or not we're allowed to submit a payment via `GET` request.
Ideally this is always `False`, but some payment providers only
support a redirect via `GET`. Make sure the token retrieval is
sufficiently secure to make up for this shortcoming.
"""
return False