gauvainboiche
94 lines
2.9 KiB
Python
94 lines
2.9 KiB
Python
"""
|
|
**Consignes :**
|
|
|
|
1. Définissez deux modèles `Equipement` et `Interface` avec une relation **1-N** (`relationship` + `ForeignKey` + `back_populates`).
|
|
|
|
2. Créez le moteur SQLite et générez les tables (`create_all`).
|
|
|
|
3. Dans une `Session`, créez **un équipement avec ses deux interfaces en une seule fois**, puis `commit`.
|
|
|
|
4. Récupérez les équipements avec `select()` et affichez, pour chacun, la liste de ses interfaces.
|
|
"""
|
|
|
|
from sqlalchemy import create_engine, select
|
|
from sqlalchemy.orm import DeclarativeMeta, declarative_base, sessionmaker
|
|
|
|
Base: DeclarativeMeta = declarative_base()
|
|
db_url = "sqlite:///exo_2_1.db"
|
|
|
|
def get_engine():
|
|
return create_engine(url= db_url)
|
|
|
|
def get_session():
|
|
engine = get_engine()
|
|
Session = sessionmaker(bind= engine)
|
|
return Session()
|
|
|
|
engine = get_engine()
|
|
|
|
from random import choice, randint
|
|
|
|
from sqlalchemy import ForeignKey, Column, Integer, Float, String, Boolean
|
|
from sqlalchemy.orm import relationship
|
|
|
|
class Equipment(Base):
|
|
__tablename__ = "equipments"
|
|
|
|
id = Column(Integer, primary_key= True, autoincrement= True)
|
|
hostname = Column(String, default= "[HOSTNAME]", nullable= False)
|
|
type = Column(String, default= "Server", nullable= False)
|
|
ip_address = Column(String, default= "255.255.255.0", nullable= False)
|
|
is_active = Column(Boolean, default= True, nullable= False)
|
|
latency = Column(Float, nullable= False)
|
|
|
|
interfaces = relationship("Interface", back_populates="equipment")
|
|
|
|
class Interface(Base):
|
|
__tablename__ = "interfaces"
|
|
|
|
id = Column(Integer, primary_key= True, autoincrement= True)
|
|
port = Column(Integer, nullable= False)
|
|
service = Column(String, default= "[SERVICE]", nullable= False)
|
|
|
|
equipment_id = Column(Integer, ForeignKey("equipments.id"), nullable= False)
|
|
|
|
equipment = relationship("Equipment", back_populates="interfaces")
|
|
|
|
Base.metadata.create_all(engine)
|
|
|
|
session = get_session()
|
|
|
|
def random_ipv4() -> str:
|
|
return ".".join(str(randint(0, 255)) for _ in range(4))
|
|
|
|
def load_words(path: str = "wordlist.txt") -> list[str]:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
return [line.strip() for line in f if line.strip()]
|
|
|
|
def random_hostname(words: list[str] | None = None) -> str:
|
|
if words is None:
|
|
words = load_words()
|
|
triplet = "-".join(choice(words) for _ in range(3))
|
|
number = f"{randint(1, 999):03d}"
|
|
return f"{triplet}-{number}"
|
|
|
|
equipment = Equipment(
|
|
hostname=random_hostname(),
|
|
type="Server",
|
|
ip_address=random_ipv4(),
|
|
is_active=True,
|
|
latency=12.5,
|
|
interfaces=[
|
|
Interface(port=22, service="SSH"),
|
|
Interface(port=443, service="HTTPS"),
|
|
],
|
|
)
|
|
|
|
session.add(equipment)
|
|
session.commit()
|
|
|
|
stmt = select(Equipment)
|
|
for eq in session.execute(stmt).scalars():
|
|
print(f"{eq.hostname} ({eq.ip_address})")
|
|
for iface in eq.interfaces:
|
|
print(f" - port {iface.port} : {iface.service}") |