Exchanges, indices and stocks imperative mapping

orm.py

import logging
from sqlalchemy import (Table, Column, Integer, Enum, Float, String, Date, Boolean, ForeignKey)
from sqlalchemy.orm import registry, relationship

from bot.domain import model

"""
Imperative mapping with dataclasses (models are dataclasses)
https://docs.sqlalchemy.org/en/14/orm/mapping_styles.html#imperative-mapping-with-dataclasses-and-attrs
"""
logger = logging.getLogger(__name__)
mapper_registry = registry()

instruments_table = Table(
    'instruments',
    mapper_registry.metadata,
    Column('symbol', String(12), primary_key=True),
    Column('name', String(200), nullable=False),
    Column('type', Enum(model.TradingInstrumentTypeEnum), nullable=False),
)

indexes_table = Table(
    'indexes',
    mapper_registry.metadata,
    Column('symbol', String(12), ForeignKey('instruments.symbol'), primary_key=True),
)

exchanges_table = Table(
    'exchanges',
    mapper_registry.metadata,
    Column('symbol', String(8), primary_key=True),
    Column('mic', String(10), unique=True, nullable=True),
    Column('acronym', String(4)),
    Column('name', String(200), nullable=True),
    Column('region', String(2)),
    Column('suffix', String(3)),
)

stocks_table = Table(
    'stocks',
    mapper_registry.metadata,
    Column('symbol', String(12), ForeignKey('instruments.symbol'), primary_key=True),
    Column('cik', String(10), nullable=True),
    Column('figi', String(12), nullable=True),
    Column('isin', String(12), nullable=True),
    Column('beta', Float, default=0.0),
    Column('market_cap', Float, default=0.0),
    Column('is_traded', Boolean, nullable=True, default=True),
    Column('next_dividend_date', Date, nullable=True),
    Column('outstanding_shares', Integer, default=0),
    Column('float_shares', Integer, default=0),
    Column('exchange_symbol', String(8), ForeignKey('exchanges.symbol'), nullable=True),
)

indexes_components_table = Table(
    'indexes_components',
    mapper_registry.metadata,
    Column('index_symbol', String(12), ForeignKey('indexes.symbol'), primary_key=True),
    Column('stock_symbol', String(12), ForeignKey('stocks.symbol'), primary_key=True),
)


def to_dict(self):
    result = {c.name: getattr(self, c.name)
              for c in self.__table__.columns}
    return result


def start_mappers():
    logger.info("Starting mappers")
    instruments_mapper = mapper_registry.map_imperatively(model.TradingInstrument, instruments_table,
                                                          polymorphic_on=instruments_table.c.type,
                                                          polymorphic_identity='INSTRUMENTS',
                                                          )

    mapper_registry.map_imperatively(model.Index, indexes_table,
                                     inherits=instruments_mapper,
                                     polymorphic_identity='INDEX'
                                     )

    mapper_registry.map_imperatively(model.Stock, stocks_table,
                                     inherits=instruments_mapper,
                                     polymorphic_identity='STOCK',
                                     properties={
                                         'exchange': relationship(model.Exchange),
                                         'indices': relationship(model.Index,
                                                                 secondary=indexes_components_table)}
                                     )

    mapper_registry.map_imperatively(model.Exchange, exchanges_table, properties={
        'instruments': relationship(model.TradingInstrument)})