from sqlalchemy import ( create_engine, UUID, Column, Integer, String, ForeignKey, Boolean, DateTime, func, text, Index, UniqueConstraint, ) from sqlalchemy.orm import relationship, declarative_base from .base import CrudMixin from sqlalchemy_mixins.serialize import SerializeMixin from sqlalchemy_mixins.repr import ReprMixin from sqlalchemy_mixins.smartquery import SmartQueryMixin Base = declarative_base() class CrudCollection(Base, CrudMixin, ReprMixin, SmartQueryMixin, SerializeMixin): __abstract__ = True __repr__ = ReprMixin.__repr__ id: int = Column(Integer, primary_key=True) uu_id: str = Column( UUID, server_default=text("gen_random_uuid()"), index=True, unique=True ) createdate = Column(DateTime, default=func.now()) updatedate = Column(DateTime, default=func.now(), onupdate=func.now()) createperson = Column(Integer, ForeignKey("persons.id"), nullable=True) updateperson = Column(Integer, ForeignKey("persons.id"), nullable=True) active = Column(Boolean, default=True) delete = Column(Boolean, default=False) replicationid = Column(Integer) class Address(CrudCollection): __tablename__ = "address" __table_args__ = ( Index("_address_ndx_00", country_code, b_state, city, district), {"comment": "Address Information"}, ) country_code = mapped_column( String(9), server_default="TR", nullable=False, comment="Country Code" ) city = mapped_column(String(24), nullable=False, comment="City") district = mapped_column(String(24), nullable=False, comment="District") b_state = mapped_column(String(24), server_default="") neighborhood = mapped_column(String(24), server_default="") street = mapped_column(String(24), server_default="") postcode = mapped_column(String(8), server_default="") latitude = mapped_column(Numeric(20, 12), server_default="0") longitude = mapped_column(Numeric(20, 12), server_default="0") official_companies: Mapped["Company"] = relationship( "Company", back_populates="official_address", foreign_keys="Company.official_address_id", ) class Company(CrudCollection): __tablename__ = "companies" __table_args__ = ( UniqueConstraint("tax_number", name="uq_companies_tax_number"), {"comment": "İşletmelerin bilgilerini tutar"}, ) official_name = mapped_column(String, nullable=True, comment="Şirketin resmi adı") public_name = Column(String, nullable=False, comment="Şirketin bilinen adı") tax_number = Column(String, nullable=False, unique=True, comment="Vergi numarası") organizations = relationship("Organization", back_populates="company") locations = relationship("Location", back_populates="company") default_lang_type = mapped_column(String(5), server_default="TR") default_money_type = mapped_column(String(5), server_default="TL") is_commercial = mapped_column(Boolean, server_default="True") is_blacklist = mapped_column(Boolean, server_default="False") official_address_id = mapped_column(ForeignKey("address.id")) parent_id = mapped_column(ForeignKey("company.id")) official_address: Mapped[List["Address"]] = relationship( "Address", back_populates="official_companies", foreign_keys=[official_address_id], ) class Organization(CrudCollection): __tablename__ = "organizations" __table_args__ = {"comment": "Her şirketin organizasyon yapısını tutar"} name = Column(String, nullable=False, comment="Organizasyonun adı") company_id = Column( Integer, ForeignKey("companies.id"), nullable=False, comment="Şirket ID" ) valid_from = Column( DateTime, default=func.now(), comment="Geçerlilik başlangıç tarihi" ) valid_to = Column(DateTime, nullable=True, comment="Geçerlilik bitiş tarihi") company = relationship("Company", back_populates="organizations") employees = relationship("Employee", back_populates="organization") locations = relationship("Location", back_populates="organization") phones = relationship("Phone", back_populates="organization") menus = relationship("Menu", back_populates="organization") class Location(CrudCollection): __tablename__ = "locations" __table_args__ = { "comment": "Şirketin farklı lokasyonlarını (depo, ofis, şube, büro vb.) tutar" } name = Column(String, nullable=False, comment="Lokasyon adı") address = Column(String, nullable=False, comment="Lokasyon adresi") location_type = Column( String, nullable=False, comment="Lokasyon türü (depo, ofis, şube, büro vb.)" ) company_id = Column( Integer, ForeignKey("companies.id"), nullable=False, comment="Şirket ID" ) organization_id = Column( Integer, ForeignKey("organizations.id"), nullable=False, comment="Organizasyon ID", ) company = relationship("Company", back_populates="locations") organization = relationship("Organization", back_populates="locations") class Person(CrudCollection): __tablename__ = "persons" __table_args__ = ( UniqueConstraint("identity_number", name="uq_persons_identity_number"), {"comment": "Kişilerin bilgilerini tutar"}, ) first_name = Column(String, nullable=False, comment="İlk isim") last_name = Column(String, nullable=False, comment="Soy isim") identity_number = Column( String, nullable=False, unique=True, comment="Kimlik numarası" ) mother_name = Column(String, nullable=False, comment="Anne adı") father_name = Column(String, nullable=False, comment="Baba adı") birth_place = Column(String, nullable=False, comment="Doğum yeri") birth_date = Column(DateTime, nullable=False, comment="Doğum tarihi") gender = Column(String, nullable=False, comment="Cinsiyet") nationality_code = Column(String, nullable=False, comment="Ülke kimlik kodu") employees = relationship("Employee", back_populates="person") users = relationship("User", back_populates="person") class Employee(CrudCollection): __tablename__ = "employees" __table_args__ = ( UniqueConstraint("person_id", name="uq_employee_person_id"), {"comment": "Çalışanların bilgilerini tutar"}, ) person_id = Column( Integer, ForeignKey("persons.id"), nullable=False, unique=True, comment="Kişi ID", ) organization_id = Column( Integer, ForeignKey("organizations.id"), nullable=False, comment="Organizasyon ID", ) person = relationship("Person", back_populates="employees") organization = relationship("Organization", back_populates="employees") processes = relationship("EmployeeProcess", back_populates="employee") permissions = relationship("Permission", back_populates="employee") salaries = relationship("Salary", back_populates="employee") performances = relationship("Performance", back_populates="employee") """ class User(CrudCollection): __tablename__ = 'users' __table_args__ = ( UniqueConstraint('email', name='uq_users_email'), UniqueConstraint('phone_number', name='uq_users_phone_number'), {'comment': 'Kullanıcıların giriş bilgilerini tutar'} ) person_id = Column(Integer, ForeignKey('persons.id'), nullable=False, comment='Kişi ID') email = Column(String, nullable=True, unique=True, comment='E-posta adresi') phone_number = Column(String, nullable=True, unique=True, comment='Telefon numarası') password_hash = Column(String, nullable=False, comment='Parola hash değeri') person = relationship('Person', back_populates='users') login_attempts = relationship('LoginAttempt', back_populates='user') class LoginAttempt(CrudCollection): __tablename__ = 'login_attempts' __table_args__ = {'comment': 'Kullanıcıların giriş denemelerini tutar'} user_id = Column(Integer, ForeignKey('users.id'), nullable=False, comment='Kullanıcı ID') timestamp = Column(DateTime, default=func.now(), nullable=False, comment='Giriş deneme zamanı') success = Column(Boolean, default=False, nullable=False, comment='Giriş başarılı mı') user = relationship('User', back_populates='login_attempts') """ class Process(CrudCollection): __tablename__ = "processes" __table_args__ = {"comment": "İş süreçlerinin tanımlamalarını tutar"} name = Column(String, nullable=False, comment="Süreç adı") employee_processes = relationship("EmployeeProcess", back_populates="process") emails = relationship("Email", back_populates="process") class EmployeeProcess(CrudCollection): __tablename__ = "employee_processes" __table_args__ = { "comment": "Çalışan ile iş süreçleri arasındaki ilişkileri ve yetkileri tutar" } employee_id = Column( Integer, ForeignKey("employees.id"), nullable=False, comment="Çalışan ID" ) process_id = Column( Integer, ForeignKey("processes.id"), nullable=False, comment="Süreç ID" ) can_add = Column(Boolean, default=False, comment="Ekleme yetkisi") can_update = Column(Boolean, default=False, comment="Güncelleme yetkisi") can_delete = Column(Boolean, default=False, comment="Silme yetkisi") employee = relationship("Employee", back_populates="processes") process = relationship("Process", back_populates="employee_processes") class Permission(CrudCollection): __tablename__ = "permissions" __table_args__ = {"comment": "Yetkileri tanımlar ve çalışanlara devredilebilir"} employee_id = Column( Integer, ForeignKey("employees.id"), nullable=False, comment="Çalışan ID" ) granted_to = Column( Integer, ForeignKey("employees.id"), nullable=True, comment="Yetki verilen çalışan ID", ) # Temporary delegation permission_type = Column( String, nullable=False, comment="Yetki türü (add, update, delete)" ) valid_until = Column(DateTime, nullable=True, comment="Geçerlilik bitiş tarihi") employee = relationship( "Employee", foreign_keys=[employee_id], back_populates="permissions" ) granted_employee = relationship("Employee", foreign_keys=[granted_to]) class Performance(CrudCollection): __tablename__ = "performances" __table_args__ = {"comment": "Çalışanların performanslarını tutar"} employee_id = Column( Integer, ForeignKey("employees.id"), nullable=False, comment="Çalışan ID" ) review_date = Column(DateTime, nullable=False, comment="Değerlendirme tarihi") score = Column(Float, nullable=False, comment="Performans puanı") notes = Column(String, nullable=True, comment="Notlar") employee = relationship("Employee", back_populates="performances") class Salary(CrudCollection): __tablename__ = "salaries" __table_args__ = {"comment": "Çalışanların maaş bilgilerini tutar"} employee_id = Column( Integer, ForeignKey("employees.id"), nullable=False, comment="Çalışan ID" ) salary_amount = Column(Float, nullable=False, comment="Maaş miktarı") effective_date = Column(DateTime, nullable=False, comment="Geçerlilik tarihi") employee = relationship("Employee", back_populates="salaries") class Email(CrudCollection): __tablename__ = "emails" __table_args__ = { "comment": "İş süreçlerine ve çalışanlara e-posta gönderimlerini takip eder" } process_id = Column( Integer, ForeignKey("processes.id"), nullable=False, comment="Süreç ID" ) recipient = Column(String, nullable=False, comment="Alıcı e-posta adresi") subject = Column(String, nullable=False, comment="E-posta konusu") body = Column(String, nullable=False, comment="E-posta içeriği") sent_date = Column( DateTime, default=func.now(), nullable=False, comment="Gönderim tarihi" ) process = relationship("Process", back_populates="emails") class Phone(CrudCollection): __tablename__ = "phones" __table_args__ = {"comment": "Organizasyonlara ait telefon bilgilerini tutar"} organization_id = Column( Integer, ForeignKey("organizations.id"), nullable=False, comment="Organizasyon ID", ) phone_number = Column(String, nullable=False, comment="Telefon numarası") phone_type = Column( String, nullable=False, comment="Telefon türü (Sabit, Mobil vb.)" ) organization = relationship("Organization", back_populates="phones") class Menu(CrudCollection): __tablename__ = "menus" __table_args__ = ( Index("ix_menus_title", "title"), {"comment": "Menü öğelerini tutar"}, ) title = Column(String, nullable=False, comment="Menü başlığı") link = Column(String, nullable=False, comment="Menü bağlantısı") icon = Column(String, nullable=True, comment="Menü simgesi") order = Column(Integer, nullable=False, default=0, comment="Menü sıralaması") organization_id = Column( Integer, ForeignKey("organizations.id"), nullable=False, comment="Organizasyon ID", ) page_id = Column(Integer, ForeignKey("pages.id"), nullable=True, comment="Sayfa ID") sub_menus = relationship( "SubMenu", back_populates="menu", cascade="all, delete-orphan" ) organization = relationship("Organization", back_populates="menus") page = relationship("Page", back_populates="menus") class SubMenu(CrudCollection): __tablename__ = "sub_menus" __table_args__ = ( Index("ix_sub_menus_title", "title"), {"comment": "Alt menü öğelerini tutar"}, ) menu_id = Column( Integer, ForeignKey("menus.id"), nullable=False, comment="Ana Menü ID" ) title = Column(String, nullable=False, comment="Alt Menü başlığı") link = Column(String, nullable=False, comment="Alt Menü bağlantısı") icon = Column(String, nullable=True, comment="Alt Menü simgesi") order = Column(Integer, nullable=False, default=0, comment="Alt Menü sıralaması") layer = Column(Integer, nullable=False, default=0, comment="Alt Menü seviyesi") page_id = Column(Integer, ForeignKey("pages.id"), nullable=True, comment="Sayfa ID") menu = relationship("Menu", back_populates="sub_menus") page = relationship("Page", back_populates="sub_menus") class Page(CrudCollection): __tablename__ = "pages" __table_args__ = {"comment": "Uygulamada kullanılacak sayfaları tutar"} title = Column(String, nullable=False, comment="Sayfa başlığı") path = Column(String, nullable=False, comment="Sayfa yolu") menus = relationship("Menu", back_populates="page") sub_menus = relationship("SubMenu", back_populates="page") functions = relationship("Function", back_populates="page") class Function(CrudCollection): __tablename__ = "functions" __table_args__ = { "comment": "Uygulamada kullanılacak fonksiyon parçacıklarını tutar" } name = Column(String, nullable=False, comment="Fonksiyon adı") description = Column(String, nullable=True, comment="Fonksiyon açıklaması") code = Column(String, nullable=False, comment="Fonksiyon kodu") page_id = Column( Integer, ForeignKey("pages.id"), nullable=False, comment="Sayfa ID" ) page = relationship("Page", back_populates="functions") from sqlalchemy.exc import NoResultFound def authenticate(email=None, phone_number=None, password=None): try: if email: user = session.query(User).filter(User.email == email).one() elif phone_number: user = session.query(User).filter(User.phone_number == phone_number).one() else: return False # Şifre doğrulama (gerçek şifre doğrulama fonksiyonu kullanılmalıdır) if user.password_hash == password: login_attempt = LoginAttempt(user_id=user.id, success=True) session.add(login_attempt) session.commit() return True else: login_attempt = LoginAttempt(user_id=user.id, success=False) session.add(login_attempt) session.commit() return False except NoResultFound: return False # PostgreSQL veritabanına bağlan engine = create_engine("postgresql://username:password@localhost/mydatabase") Base.metadata.create_all(engine) # Oturum oluştur Session = sessionmaker(bind=engine) session = Session() # Örnek veriler ekle company = Company(name="My Company", tax_number="1234567890") session.add(company) session.commit() organization = Organization(name="IT Department", company_id=company.id) session.add(organization) session.commit() location = Location( name="Main Office", address="123 Main St", location_type="Office", company_id=company.id, organization_id=organization.id, ) session.add(location) session.commit() person = Person( first_name="John", last_name="Doe", identity_number="9876543210", mother_name="Jane Doe", father_name="Joe Doe", birth_place="New York", birth_date="1980-01-01", gender="Male", nationality_code="US", ) session.add(person) session.commit() employee = Employee(person_id=person.id, organization_id=organization.id) session.add(employee) session.commit() user = User( person_id=person.id, email="john.doe@example.com", phone_number="555-1234", password_hash="hashed_password", ) session.add(user) session.commit() login_attempt = LoginAttempt(user_id=user.id, success=True) session.add(login_attempt) session.commit() process = Process(name="Code Review") session.add(process) session.commit() employee_process = EmployeeProcess( employee_id=employee.id, process_id=process.id, can_add=True, can_update=True, can_delete=False, ) session.add(employee_process) session.commit() permission = Permission( employee_id=employee.id, permission_type="add", valid_until=None ) session.add(permission) session.commit() performance = Performance(employee_id=employee.id, review_date="2023-01-01", score=4.5) session.add(performance) session.commit() salary = Salary( employee_id=employee.id, amount=5000.0, effective_from="2023-01-01", effective_to="2024-01-01", ) session.add(salary) session.commit() phone1 = Phone( organization_id=organization.id, phone_number="555-1234", phone_type="mobil" ) session.add(phone1) phone2 = Phone( organization_id=organization.id, phone_number="555-5678", phone_type="sabit" ) session.add(phone2) session.commit() email = Email(process_id=process.id, email_address="codereview@example.com") session.add(email) session.commit()