headless_lms_server/domain/csv_export/
mod.rs

1pub mod code_giveaway_codes;
2pub mod course_instance_export;
3pub mod course_research_form_questions_answers_export;
4pub mod credit_registrations_export;
5pub mod exercise_tasks_export;
6pub mod points;
7pub mod submissions;
8pub mod user_exercise_states_export;
9pub mod users_export;
10
11use anyhow::{Context, Result};
12use bytes::Bytes;
13use csv::Writer;
14use futures::{Stream, StreamExt, stream::FuturesUnordered};
15
16use async_trait::async_trait;
17
18use models::course_module_completions::CourseModuleCompletionWithRegistrationInfo;
19use serde::Serialize;
20use sqlx::PgConnection;
21use std::{
22    io,
23    io::Write,
24    sync::{
25        Arc, Mutex,
26        atomic::{AtomicBool, Ordering},
27    },
28};
29use tokio::{sync::mpsc::UnboundedSender, task::JoinHandle};
30use tokio_stream::wrappers::UnboundedReceiverStream;
31
32use crate::prelude::*;
33
34use super::authorization::{AuthorizationToken, AuthorizedResponse};
35/// Convenience struct for creating CSV data.
36struct CsvWriter<W: Write> {
37    csv_writer: Arc<Mutex<Writer<W>>>,
38    handles: FuturesUnordered<JoinHandle<Result<()>>>,
39}
40
41impl<W: Write + Send + 'static> CsvWriter<W> {
42    /// Creates a new CsvWriter, and also writes the given headers before returning.
43    async fn new_with_initialized_headers<I, T>(writer: W, headers: I) -> Result<Self>
44    where
45        I: IntoIterator<Item = T> + Send + 'static,
46        T: AsRef<[u8]>,
47    {
48        let mut writer = csv::WriterBuilder::new()
49            .has_headers(false)
50            .from_writer(writer);
51
52        // write headers first
53        let writer = tokio::task::spawn_blocking(move || {
54            writer
55                .write_record(headers)
56                .context("Failed to write headers")?;
57            Result::<_, anyhow::Error>::Ok(writer)
58        })
59        .await??;
60
61        Ok(Self {
62            csv_writer: Arc::new(Mutex::new(writer)),
63            handles: FuturesUnordered::new(),
64        })
65    }
66
67    /// Spawns a task that writes a single CSV record
68    fn write_record<I, T>(&self, csv_row: I)
69    where
70        I: IntoIterator<Item = T> + Send + 'static,
71        T: AsRef<[u8]>,
72    {
73        let writer = self.csv_writer.clone();
74        let handle = tokio::task::spawn_blocking(move || {
75            writer
76                .lock()
77                .map_err(|_| anyhow::anyhow!("Failed to lock mutex"))?
78                .write_record(csv_row)
79                .context("Failed to serialize points")
80        });
81        self.handles.push(handle);
82    }
83
84    /// Waits for handles to finish, flushes the writer and extracts the inner writer.
85    /// Should always be called before dropping the writer to make sure writing the CSV finishes properly.
86    async fn finish(mut self) -> Result<W> {
87        // ensure every task is finished before the writer is extracted
88        while let Some(handle) = self.handles.next().await {
89            handle??;
90        }
91
92        let writer = tokio::task::spawn_blocking(move || {
93            let _ = &self;
94            Arc::try_unwrap(self.csv_writer)
95                .map_err(|_| anyhow::anyhow!("Failed to extract inner writer from arc"))?
96                .into_inner()
97                .map_err(|e| anyhow::anyhow!("Failed to extract inner writer from mutex: {}", e))?
98                .into_inner()
99                .map_err(|e| {
100                    anyhow::anyhow!("Failed to extract inner writer from CSV writer: {}", e)
101                })
102        })
103        .await??;
104        Ok(writer)
105    }
106}
107
108/**
109 * For csv export. Return the grade as a number if there is a numeric grade. If the grade is not numeric, returns pass/fail/
110 * If course module has not been completed yet, returns "-".
111 */
112fn course_module_completion_info_to_grade_string(
113    input: Option<&CourseModuleCompletionWithRegistrationInfo>,
114) -> String {
115    let grade_string = input.map(|info| {
116        if let Some(grade) = info.grade {
117            return grade.to_string();
118        }
119        if info.passed {
120            return "pass".to_string();
121        };
122        "fail".to_string()
123    });
124    if let Some(grade_string) = grade_string {
125        if let Some(info) = input
126            && !info.prerequisite_modules_completed
127        {
128            return format!("{grade_string}*");
129        }
130        return grade_string;
131    }
132    "-".to_string()
133}
134
135pub struct CSVExportAdapter {
136    pub sender: UnboundedSender<ControllerResult<Bytes>>,
137    pub authorization_token: AuthorizationToken,
138}
139impl Write for CSVExportAdapter {
140    fn flush(&mut self) -> std::io::Result<()> {
141        Ok(())
142    }
143
144    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
145        let bytes = Bytes::copy_from_slice(buf);
146        let token = self.authorization_token;
147        self.sender
148            .send(token.authorized_ok(bytes))
149            .map_err(|e| io::Error::other(e.to_string()))?;
150        Ok(buf.len())
151    }
152}
153
154/** Without this one, actix cannot stream our authorized streams as responses
155
156```ignore
157HttpResponse::Ok()
158    .append_header((
159        "Content-Disposition",
160        format!(
161            "attachment; filename=\"Exam: {} - Submissions {}.csv\"",
162            exam.name,
163            Utc::now().format("%Y-%m-%d")
164        ),
165    ))
166    .streaming(make_authorized_streamable(UnboundedReceiverStream::new(
167        receiver,
168    ))),
169```
170*/
171pub fn make_authorized_streamable(
172    stream: impl Stream<Item = Result<AuthorizedResponse<bytes::Bytes>, ControllerError>>,
173) -> impl Stream<Item = Result<bytes::Bytes, ControllerError>> {
174    stream.map(|item| item.map(|item2| item2.data))
175}
176
177/**
178  For streaming arrays of json objects.
179*/
180pub fn serializable_sqlx_result_stream_to_json_stream(
181    stream: impl Stream<Item = sqlx::Result<impl Serialize>>,
182) -> impl Stream<Item = Result<bytes::Bytes, ControllerError>> {
183    // The opening bracket rides along with the first item, so the terminator has to know whether
184    // one was ever emitted; without this an empty stream would answer with a bare `]`.
185    let emitted_any_item = Arc::new(AtomicBool::new(false));
186    let mark_emitted = Arc::clone(&emitted_any_item);
187    let res_stream = stream.enumerate().map(move |(n, item)| {
188        mark_emitted.store(true, Ordering::Relaxed);
189        item.map(|item2| {
190            match serde_json::to_vec(&item2) {
191                Ok(mut v) => {
192                    // Only item index available, we don't know the length of the stream
193                    if n == 0 {
194                        // Start of array character to the beginning of the stream
195                        v.insert(0, b'[');
196                    } else {
197                        // Separator character before every item excluding the first item
198                        v.insert(0, b',');
199                    }
200                    Bytes::from(v)
201                }
202                Err(e) => {
203                    // Since we're already streaming a response, we have no way to change the status code of the response anymore.
204                    // Our best option at this point is to write the error to the response, hopefully causing the response to be invalid json.
205                    error!("Failed to serialize item: {}", e);
206                    Bytes::from(format!(
207                        "Streaming error: Failed to serialize item. Details: {:?}",
208                        e
209                    ))
210                }
211            }
212        })
213        .map_err(|original_error| {
214            ControllerError::new(
215                ControllerErrorType::InternalServerError,
216                original_error.to_string(),
217                Some(original_error.into()),
218            )
219        })
220    });
221    // Chaining the end of the json array character here because in the previous map we don't know the length of the stream
222    res_stream.chain(futures::stream::once(async move {
223        Ok(Bytes::from_static(
224            if emitted_any_item.load(Ordering::Relaxed) {
225                b"]"
226            } else {
227                b"[]"
228            },
229        ))
230    }))
231}
232
233#[async_trait]
234pub trait CsvExportDataLoader {
235    async fn load_data(
236        &self,
237        sender: UnboundedSender<Result<AuthorizedResponse<Bytes>, ControllerError>>,
238        conn: &mut PgConnection,
239        token: AuthorizationToken,
240    ) -> anyhow::Result<CSVExportAdapter>;
241}
242
243pub async fn general_export(
244    pool: web::Data<PgPool>,
245    content_disposition: &str,
246    data_loader: impl CsvExportDataLoader + std::marker::Send + 'static,
247    token: AuthorizationToken,
248) -> ControllerResult<HttpResponse> {
249    let (sender, receiver) = tokio::sync::mpsc::unbounded_channel::<ControllerResult<Bytes>>();
250    // spawn handle that writes the csv row by row into the sender
251    let mut handle_conn = pool.acquire().await?;
252    let _handle = tokio::spawn(async move {
253        let fut = data_loader.load_data(sender, &mut handle_conn, token);
254        let res = fut.await;
255        if let Err(err) = res {
256            tracing::error!("Failed to export: {}", err);
257        }
258    });
259
260    // return response that streams data from the receiver
261    token.authorized_ok(
262        HttpResponse::Ok()
263            .append_header(("Content-Disposition", content_disposition))
264            .streaming(make_authorized_streamable(UnboundedReceiverStream::new(
265                receiver,
266            ))),
267    )
268}
269
270#[cfg(test)]
271mod test {
272    use std::{collections::HashMap, io::Cursor};
273
274    use headless_lms_models::{
275        course_instance_enrollments, exercise_slides,
276        exercise_task_gradings::ExerciseTaskGradingResult,
277        exercise_tasks::{self, NewExerciseTask},
278        exercises::{self, GradingProgress},
279        library::grading::{
280            GradingPolicy, StudentExerciseSlideSubmission, StudentExerciseTaskSubmission,
281        },
282        user_exercise_states,
283        user_exercise_states::ExerciseWithUserState,
284        users,
285    };
286    use models::chapters::{self, NewChapter};
287    use serde_json::Value;
288
289    use super::*;
290    use crate::{
291        domain::{
292            csv_export::points::export_course_instance_points,
293            models_requests::{self, JwtKey},
294        },
295        test_helper::*,
296    };
297
298    #[actix_web::test]
299    async fn exports() {
300        insert_data!(:tx, :user, :org, :course, :instance, :course_module, :chapter, :page, :exercise, :slide, :task);
301
302        let u2 = users::insert(
303            tx.as_mut(),
304            PKeyPolicy::Generate,
305            "second@example.org",
306            None,
307            None,
308        )
309        .await
310        .unwrap();
311
312        course_instance_enrollments::insert(tx.as_mut(), user, course, instance.id)
313            .await
314            .unwrap();
315        course_instance_enrollments::insert(tx.as_mut(), u2, course, instance.id)
316            .await
317            .unwrap();
318
319        let c2 = chapters::insert(
320            tx.as_mut(),
321            PKeyPolicy::Generate,
322            &NewChapter {
323                name: "".to_string(),
324                color: Some("#065853".to_string()),
325                course_id: course,
326                chapter_number: 2,
327                front_page_id: None,
328                opens_at: None,
329                deadline: None,
330                course_module_id: Some(course_module.id),
331            },
332        )
333        .await
334        .unwrap();
335
336        let e2 = exercises::insert(tx.as_mut(), PKeyPolicy::Generate, course, "", page, c2, 0)
337            .await
338            .unwrap();
339        let s2 = exercise_slides::insert(tx.as_mut(), PKeyPolicy::Generate, e2, 0)
340            .await
341            .unwrap();
342        let et2 = exercise_tasks::insert(
343            tx.as_mut(),
344            PKeyPolicy::Generate,
345            NewExerciseTask {
346                exercise_slide_id: s2,
347                exercise_type: "".to_string(),
348                assignment: vec![],
349                public_spec: Some(Value::Null),
350                private_spec: Some(Value::Null),
351                model_solution_spec: Some(Value::Null),
352                order_number: 1,
353            },
354        )
355        .await
356        .unwrap();
357
358        let e3 = exercises::insert(tx.as_mut(), PKeyPolicy::Generate, course, "", page, c2, 1)
359            .await
360            .unwrap();
361        let s3 = exercise_slides::insert(tx.as_mut(), PKeyPolicy::Generate, e3, 0)
362            .await
363            .unwrap();
364        let et3 = exercise_tasks::insert(
365            tx.as_mut(),
366            PKeyPolicy::Generate,
367            NewExerciseTask {
368                exercise_slide_id: s3,
369                exercise_type: "".to_string(),
370                assignment: vec![],
371                public_spec: Some(Value::Null),
372                private_spec: Some(Value::Null),
373                model_solution_spec: Some(Value::Null),
374                order_number: 2,
375            },
376        )
377        .await
378        .unwrap();
379        submit_and_grade(tx.as_mut(), exercise, slide, task, user, course, 12.34).await;
380        submit_and_grade(tx.as_mut(), e2, s2, et2, user, course, 23.45).await;
381        submit_and_grade(tx.as_mut(), e2, s2, et2, u2, course, 34.56).await;
382        submit_and_grade(tx.as_mut(), e3, s3, et3, u2, course, 45.67).await;
383
384        let buf = vec![];
385        let buf = export_course_instance_points(tx.as_mut(), instance.id, buf)
386            .await
387            .unwrap();
388        let buf = Cursor::new(buf);
389
390        let mut reader = csv::Reader::from_reader(buf);
391        let mut count = 0;
392        for record in reader.records() {
393            count += 1;
394            let record = record.unwrap();
395            println!("{}", record.as_slice());
396            let user_id = Uuid::parse_str(&record[0]).unwrap();
397            let first = record[1].parse::<f32>().unwrap();
398            let second = record[2].parse::<f32>().unwrap();
399            if user_id == user {
400                assert!((first - 0.1234).abs() < 0.1 && (second - 0.2345).abs() < 0.1);
401            } else if user_id == u2 {
402                assert!((first - 0.0).abs() < 0.1 && (second - 0.8023).abs() < 0.1);
403            } else {
404                panic!("unexpected user id");
405            }
406        }
407        assert_eq!(count, 2)
408    }
409
410    async fn submit_and_grade(
411        tx: &mut PgConnection,
412        ex: Uuid,
413        ex_slide: Uuid,
414        et: Uuid,
415        u: Uuid,
416        course_id: Uuid,
417        score_given: f32,
418    ) {
419        let exercise = exercises::get_by_id(tx, ex).await.unwrap();
420        user_exercise_states::get_or_create_user_exercise_state(tx, u, ex, Some(course_id), None)
421            .await
422            .unwrap();
423        user_exercise_states::upsert_selected_exercise_slide_id(
424            tx,
425            u,
426            ex,
427            Some(course_id),
428            None,
429            Some(ex_slide),
430        )
431        .await
432        .unwrap();
433        let user_exercise_state = user_exercise_states::get_or_create_user_exercise_state(
434            tx,
435            u,
436            ex,
437            Some(course_id),
438            None,
439        )
440        .await
441        .unwrap();
442        let mut exercise_with_user_state =
443            ExerciseWithUserState::new(exercise, user_exercise_state).unwrap();
444        let jwt_key = Arc::new(JwtKey::test_key());
445        let app_conf = crate::test_helper::init_app_conf().expect("app conf");
446        headless_lms_models::library::grading::grade_user_submission(
447            tx,
448            &mut exercise_with_user_state,
449            &StudentExerciseSlideSubmission {
450                exercise_slide_id: ex_slide,
451                exercise_task_submissions: vec![StudentExerciseTaskSubmission::json(
452                    et,
453                    Value::Null,
454                )],
455            },
456            GradingPolicy::Fixed(HashMap::from([(
457                et,
458                ExerciseTaskGradingResult {
459                    feedback_json: None,
460                    feedback_text: None,
461                    grading_progress: GradingProgress::FullyGraded,
462                    score_given,
463                    score_maximum: 100,
464                    set_user_variables: Some(HashMap::new()),
465                },
466            )])),
467            models_requests::fetch_service_info,
468            models_requests::make_grading_request_sender(jwt_key, app_conf.base_url.clone()),
469            &crate::test_helper::init_file_store(),
470            &app_conf,
471        )
472        .await
473        .unwrap();
474    }
475}
headless_lms_server/domain/csv_export/mod.rs

headless_lms_server/domain/csv_export/
mod.rs
