Maxim Snesarev 33d6bb7ebd Refactor AMO CRM Data Collection Service to use PostgreSQL
- Updated database configuration to switch from SQLite to PostgreSQL, including changes to alembic.ini, Docker Compose, and environment settings.
- Refactored application code to utilize PostgreSQL database adapters, ensuring compatibility with the new database structure.
- Enhanced API routes and data handling to support the new database, including adjustments in data models and query logic.
- Introduced new job processing mechanisms for full synchronization of AMO CRM entities, leveraging FastStream for background tasks.
- Improved logging and error handling across the application to facilitate better monitoring and debugging.
- Removed obsolete SQLite adapter files and migrations, streamlining the project structure for PostgreSQL integration.
2025-11-05 00:38:37 +03:00

290 lines
11 KiB
Python

from sqlalchemy import (
Column, Integer, String, Boolean, Text, ForeignKey, JSON,
Table, Index, CheckConstraint, ForeignKeyConstraint
)
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from .database import Base
# Association tables for many-to-many relationships
deal_contacts = Table(
'amo_deal_contacts',
Base.metadata,
Column('deal_id', Integer, ForeignKey('amo_deals.id'), primary_key=True),
Column('contact_id', Integer, ForeignKey('amo_contacts.id'), primary_key=True),
Column('is_main', Boolean, default=False),
)
deal_companies = Table(
'amo_deal_companies',
Base.metadata,
Column('deal_id', Integer, ForeignKey('amo_deals.id'), primary_key=True),
Column('company_id', Integer, ForeignKey('amo_companies.id'), primary_key=True),
Column('is_main', Boolean, default=False),
)
contact_companies = Table(
'amo_contact_companies',
Base.metadata,
Column('contact_id', Integer, ForeignKey('amo_contacts.id'), primary_key=True),
Column('company_id', Integer, ForeignKey('amo_companies.id'), primary_key=True),
Column('is_main', Boolean, default=False),
)
class User(Base):
__tablename__ = 'amo_users'
id = Column(Integer, primary_key=True, autoincrement=False)
name = Column(String(255), nullable=False)
email = Column(String(255))
is_active = Column(Boolean, default=True)
created_at = Column(Integer)
updated_at = Column(Integer)
raw_data = Column(JSON)
# Relationships
created_deals = relationship("Deal", foreign_keys="Deal.created_by", back_populates="creator")
updated_deals = relationship("Deal", foreign_keys="Deal.updated_by", back_populates="updater")
responsible_deals = relationship("Deal", foreign_keys="Deal.responsible_user_id", back_populates="responsible_user")
created_contacts = relationship("Contact", foreign_keys="Contact.created_by", back_populates="creator")
updated_contacts = relationship("Contact", foreign_keys="Contact.updated_by", back_populates="updater")
responsible_contacts = relationship("Contact", foreign_keys="Contact.responsible_user_id", back_populates="responsible_user")
created_companies = relationship("Company", foreign_keys="Company.created_by", back_populates="creator")
updated_companies = relationship("Company", foreign_keys="Company.updated_by", back_populates="updater")
responsible_companies = relationship("Company", foreign_keys="Company.responsible_user_id", back_populates="responsible_user")
events = relationship("Event", back_populates="created_by_user")
class Pipeline(Base):
__tablename__ = 'amo_pipelines'
id = Column(Integer, primary_key=True, autoincrement=False)
name = Column(String(255), nullable=False)
sort = Column(Integer)
is_main = Column(Boolean, default=False)
is_unsorted = Column(Boolean, default=False)
is_archive = Column(Boolean, default=False)
account_id = Column(Integer)
created_at = Column(Integer)
updated_at = Column(Integer)
raw_data = Column(JSON)
# Relationships
stages = relationship("PipelineStage", back_populates="pipeline")
deals = relationship("Deal", back_populates="pipeline")
class PipelineStage(Base):
__tablename__ = 'amo_pipeline_stages'
id = Column(Integer, primary_key=True, autoincrement=False)
pipeline_id = Column(Integer, ForeignKey('amo_pipelines.id'), primary_key=True, nullable=False)
name = Column(String(255), nullable=False)
sort = Column(Integer)
is_editable = Column(Boolean, default=True)
color = Column(String(7)) # Hex color code
created_at = Column(Integer)
updated_at = Column(Integer)
raw_data = Column(JSON)
# Relationships
pipeline = relationship("Pipeline", back_populates="stages")
deals = relationship("Deal", back_populates="status", foreign_keys="[Deal.status_id, Deal.pipeline_id]")
class Company(Base):
__tablename__ = 'amo_companies'
id = Column(Integer, primary_key=True, autoincrement=False)
name = Column(String(255), nullable=False)
responsible_user_id = Column(Integer, ForeignKey('amo_users.id'))
group_id = Column(Integer)
created_by = Column(Integer, ForeignKey('amo_users.id'))
updated_by = Column(Integer, ForeignKey('amo_users.id'))
created_at = Column(Integer)
updated_at = Column(Integer)
closest_task_at = Column(Integer)
is_deleted = Column(Boolean, default=False)
raw_data = Column(JSON)
# Relationships
responsible_user = relationship("User", foreign_keys=[responsible_user_id], back_populates="responsible_companies")
creator = relationship("User", foreign_keys=[created_by], back_populates="created_companies")
updater = relationship("User", foreign_keys=[updated_by], back_populates="updated_companies")
# Many-to-many relationships
deals = relationship("Deal", secondary=deal_companies, back_populates="companies")
contacts = relationship("Contact", secondary=contact_companies, back_populates="companies")
class Contact(Base):
__tablename__ = 'amo_contacts'
id = Column(Integer, primary_key=True, autoincrement=False)
name = Column(String(255), nullable=False)
first_name = Column(String(255))
last_name = Column(String(255))
responsible_user_id = Column(Integer, ForeignKey('amo_users.id'))
group_id = Column(Integer)
created_by = Column(Integer, ForeignKey('amo_users.id'))
updated_by = Column(Integer, ForeignKey('amo_users.id'))
created_at = Column(Integer)
updated_at = Column(Integer)
closest_task_at = Column(Integer)
is_deleted = Column(Boolean, default=False)
raw_data = Column(JSON)
# Relationships
responsible_user = relationship("User", foreign_keys=[responsible_user_id], back_populates="responsible_contacts")
creator = relationship("User", foreign_keys=[created_by], back_populates="created_contacts")
updater = relationship("User", foreign_keys=[updated_by], back_populates="updated_contacts")
# Many-to-many relationships
deals = relationship("Deal", secondary=deal_contacts, back_populates="contacts")
companies = relationship("Company", secondary=contact_companies, back_populates="contacts")
class Deal(Base):
__tablename__ = 'amo_deals'
id = Column(Integer, primary_key=True, autoincrement=False)
name = Column(String(255), nullable=False)
price = Column(Integer, default=0)
responsible_user_id = Column(Integer, ForeignKey('amo_users.id'))
group_id = Column(Integer)
status_id = Column(Integer)
pipeline_id = Column(Integer, ForeignKey('amo_pipelines.id'))
loss_reason_id = Column(Integer)
created_by = Column(Integer, ForeignKey('amo_users.id'))
updated_by = Column(Integer, ForeignKey('amo_users.id'))
closed_at = Column(Integer)
created_at = Column(Integer)
updated_at = Column(Integer)
closest_task_at = Column(Integer)
is_deleted = Column(Boolean, default=False)
raw_data = Column(JSON)
# Table args for composite foreign key
__table_args__ = (
ForeignKeyConstraint(
['status_id', 'pipeline_id'],
['amo_pipeline_stages.id', 'amo_pipeline_stages.pipeline_id']
),
)
# Relationships
responsible_user = relationship("User", foreign_keys=[responsible_user_id], back_populates="responsible_deals")
creator = relationship("User", foreign_keys=[created_by], back_populates="created_deals")
updater = relationship("User", foreign_keys=[updated_by], back_populates="updated_deals")
status = relationship("PipelineStage", foreign_keys=[status_id, pipeline_id], back_populates="deals")
pipeline = relationship("Pipeline", foreign_keys=[pipeline_id], back_populates="deals")
# Many-to-many relationships
contacts = relationship("Contact", secondary=deal_contacts, back_populates="deals")
companies = relationship("Company", secondary=deal_companies, back_populates="deals")
class Event(Base):
__tablename__ = 'amo_events'
__table_args__ = (
CheckConstraint(
"type IN ('incoming_call', 'outgoing_call', 'lead_status_changed')",
name='check_event_type'
),
)
id = Column(String(26), primary_key=True) # ULID format
type = Column(String(50), nullable=False)
entity_id = Column(Integer)
entity_type = Column(String(50))
created_by = Column(Integer, ForeignKey('amo_users.id'))
created_at = Column(Integer)
value_after = Column(JSON)
value_before = Column(JSON)
account_id = Column(Integer)
raw_data = Column(JSON)
# Relationships
created_by_user = relationship("User", back_populates="events")
class CustomField(Base):
__tablename__ = 'amo_custom_fields'
id = Column(Integer, primary_key=True, autoincrement=True)
entity_type = Column(String(50), nullable=False) # 'deals', 'contacts', 'companies'
entity_id = Column(Integer, nullable=False)
field_id = Column(Integer, nullable=False)
field_name = Column(String(255), nullable=False)
field_type = Column(String(50), nullable=False) # 'text', 'numeric', 'checkbox', 'select', etc.
field_value = Column(Text)
field_value_numeric = Column(Integer)
field_value_date = Column(Integer)
is_custom = Column(Boolean, default=True)
created_at = Column(Integer)
updated_at = Column(Integer)
# Indexes
__table_args__ = (
Index('idx_custom_fields_entity', 'entity_type', 'entity_id'),
Index('idx_custom_fields_field', 'field_id'),
Index('idx_custom_fields_name', 'field_name'),
)
# Export configuration models
class ExportConfiguration(Base):
__tablename__ = 'export_configuration'
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(255), nullable=False)
sheet_id = Column(String(255), nullable=False)
date_range_start = Column(Integer)
date_range_end = Column(Integer)
is_active = Column(Boolean, default=True)
created_at = Column(Integer)
updated_at = Column(Integer)
# Relationships
entity_mappings = relationship("ExportEntityMapping", back_populates="configuration")
export_jobs = relationship("ExportJob", back_populates="configuration")
class ExportEntityMapping(Base):
__tablename__ = 'export_entity_mappings'
id = Column(Integer, primary_key=True, autoincrement=True)
configuration_id = Column(Integer, ForeignKey('export_configuration.id'), nullable=False)
entity_type = Column(String(50), nullable=False)
sheet_name = Column(String(255), nullable=False)
field_mapping = Column(JSON, nullable=False)
is_enabled = Column(Boolean, default=True)
created_at = Column(Integer)
updated_at = Column(Integer)
# Relationships
configuration = relationship("ExportConfiguration", back_populates="entity_mappings")
class ExportJob(Base):
__tablename__ = 'export_jobs'
id = Column(Integer, primary_key=True, autoincrement=True)
configuration_id = Column(Integer, ForeignKey('export_configuration.id'), nullable=False)
status = Column(String(50), nullable=False, default='pending')
records_processed = Column(Integer, default=0)
total_records = Column(Integer, default=0)
error_message = Column(Text)
started_at = Column(Integer)
completed_at = Column(Integer)
created_at = Column(Integer)
# Relationships
configuration = relationship("ExportConfiguration", back_populates="export_jobs")