from pendulum import DateTime

from airflow import DAG
from airflow.decorators import task

with DAG(...) as dag:

    @task()
    def sender1():
        # 'data' key ostida brand qiymatini yuborish
        ti.xcom_push(key='data', value={'brand': 'APPLE'})

from airflow import DAG
from airflow.decorators import task

with DAG(...) as dag:

    @task()
    def sender2():
        # 'return_value' keyi avtomatik tarzda ishlatiladi
        return {'brand': 'Apple'}

from airflow import DAG
from airflow.decorators import task

with DAG(...) as dag:

    @task()
    def receiver1():
        # sender task'dan 'data' keyidagi qiymatni olish
        ti.xcom_pull(task_ids='sender', key='data')

sender() >> receiver()  

receiver(sender())  

PythonOperator(
    task_id='process_data',
    python_callable=process_function
)

BashOperator(
    task_id='print_date',
    bash_command='date'
)

DummyOperator(task_id='start')

BranchPythonOperator(
    task_id='check_type',
    python_callable=check_user_type
)

ShortCircuitOperator(
    task_id='continue_if_true',
    python_callable=check_condition
)

EmailOperator(
    task_id='send_email',
    to='you@example.com',
    subject='Task Done',
    html_content='Task finished'
)

TriggerDagRunOperator(
    task_id='trigger_other_dag',
    trigger_dag_id='another_dag'
)

FileSensor(
    task_id='wait_for_file',
    filepath='/data/file.csv',
    poke_interval=60
)

from airflow.providers.docker.operators.docker import DockerOperator

run_etl = DockerOperator(
    task_id='run_etl',
    image='myrepo/etl:1.2.0',
    command='python etl.py --date {{ ds }}',
    auto_remove=True,      
    mounts=['/opt/airflow/data:/data'],
    mem_limit='1g',
    dag=dag
)

from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
train = KubernetesPodOperator(
    task_id='train_model',
    name='train-model-{{ ds_nodash }}',
    namespace='airflow',
    image='myrepo/ml-train:1.0.0',
    cmds=['python', 'train.py'],
    arguments=['--epochs', '20'],
    resources={'request_memory': '2Gi', 'request_cpu': '1000m'},
    get_logs=True,
    is_delete_operator_pod=True
)

TRUNCATE TABLE customers;
INSERT INTO customers
SELECT * FROM staging.customers;

-- Oxirgi yuklangan sana saqlanadi
SELECT * FROM source_table
WHERE updated_at > '2025-04-27 23:59:59';

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("streaming_app").getOrCreate()
df = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "localhost:9092").option("subscribe", "topic_name").load()

from airflow import DAG
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.providers.microsoft.mssql.operators.mssql import MsSqlOperator
from datetime import datetime

with DAG(dag_id="full_load_customers",
         start_date=datetime(2024, 1, 1),
         schedule_interval="@daily",
         catchup=False) as dag:

    truncate_target_table = MsSqlOperator(
        task_id='truncate_target_table',
        mssql_conn_id='mssql_target',
        sql="TRUNCATE TABLE customers;"
    )

    load_full_data = PostgresOperator(
        task_id='load_full_data',
        postgres_conn_id='postgres_source',
        sql="""
            INSERT INTO mssql_target.dbo.customers
            SELECT * FROM postgres_source.public.customers;
        """
    )

    truncate_target_table >> load_full_data